BUG #7534: walreceiver takes long time to detect n/w breakdown

*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 62,67 **** walrcv_connect_type walrcv_connect = NULL;
--- 62,69 ----
  walrcv_receive_type walrcv_receive = NULL;
  walrcv_send_type walrcv_send = NULL;
  walrcv_disconnect_type walrcv_disconnect = NULL;
+ int			wal_receiver_replication_timeout = 60 * 1000;	/* maximum time to receive one
+ 												 * WAL data message */
  
  #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
***************
*** 174,179 **** WalReceiverMain(void)
--- 176,184 ----
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
  
+ 	TimestampTz last_recv_timestamp;
+ 	TimestampTz timeout = 0;
+ 
  	/*
  	 * WalRcv should be set up already (if we are a backend, we inherit this
  	 * by fork() or EXEC_BACKEND mechanism from the postmaster).
***************
*** 282,287 **** WalReceiverMain(void)
--- 287,295 ----
  	MemSet(&reply_message, 0, sizeof(reply_message));
  	MemSet(&feedback_message, 0, sizeof(feedback_message));
  
+ 	/* Initialize the last recv timestamp */
+ 	last_recv_timestamp = GetCurrentTimestamp();
+ 
  	/* Loop until end-of-streaming or error */
  	for (;;)
  	{
***************
*** 316,327 **** WalReceiverMain(void)
--- 324,343 ----
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
+ 			/* Something is received from master, so reset last receive time*/
+ 			last_recv_timestamp = GetCurrentTimestamp();
+ 			
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
+ 			{
+ 				/* Something is received from master, so reset last receive time*/
+ 				last_recv_timestamp = GetCurrentTimestamp();
+ 				
  				XLogWalRcvProcessMsg(type, buf, len);
+ 			}
  
  			/* Let the master know that we received some data. */
  			XLogWalRcvSendReply();
***************
*** 334,339 **** WalReceiverMain(void)
--- 350,369 ----
  		}
  		else
  		{
+ 			/* Check if time since last receive from standby has reached the configured limit
+ 			 * No need to check if it is disabled by giving value as 0*/
+ 			if (wal_receiver_replication_timeout > 0)
+ 			{
+ 				timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
+ 														  wal_receiver_replication_timeout);
+ 
+ 				if (GetCurrentTimestamp() >= timeout)
+ 				{
+ 					ereport(ERROR,
+ 						(errmsg("Could not receive any message from WalSender for configured timeout period")));
+ 				}
+ 			}
+ 		
  			/*
  			 * We didn't receive anything new, but send a status update to the
  			 * master anyway, to report any progress in applying WAL.
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 2382,2387 **** static struct config_int ConfigureNamesInt[] =
--- 2382,2398 ----
  		NULL, NULL, NULL
  	},
  
+ 	{
+ 		{"wal_receiver_replication_timeout", PGC_SIGHUP, REPLICATION_STANDBY,
+ 			gettext_noop("Sets the maximum wait time to receive data from master."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&wal_receiver_replication_timeout,
+ 		60 * 1000, 0, INT_MAX,
+ 		NULL, NULL, NULL
+ 	},
+ 		
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 237,242 ****
--- 237,244 ----
  					# 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
+ #wal_receiver_replication_timeout = 60s	# in milliseconds; 0 disables; time 
+ 					# till receiver waits for communication from master.
  
  
  #------------------------------------------------------------------------------
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 19,24 ****
--- 19,25 ----
  
  extern int	wal_receiver_status_interval;
  extern bool hot_standby_feedback;
+ extern int wal_receiver_replication_timeout;
  
  /*
   * MAXCONNINFO: maximum size of a connection string.

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 2175,2180 **** SET ENABLE_SEQSCAN TO OFF;
--- 2175,2204 ----
         </para>
        </listitem>
       </varlistentry>
+ 	 
+ 	 <varlistentry id="guc-wal-receiver-replication-timeout" xreflabel="wal_receiver_replication_timeout">
+       <term><varname>wal_receiver_replication_timeout</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>wal_receiver_replication_timeout</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
+        <para>
+         Terminate replication connections that are inactive longer
+         than the specified number of milliseconds. This is useful for
+         the receiving standby server to detect a primary node crash or network outage.
+         A value of zero disables the timeout mechanism.  This parameter
+         can only be set in
+         the <filename>postgresql.conf</> file or on the server command line.
+         The default value is 60 seconds.
+        </para>
+        <para>
+         To prevent connections from being terminated prematurely,
+         <xref linkend="guc-wal-send-status-interval">
+         must be enabled on the primary, and its value must be less than the
+         value of <varname>wal_receiver_replication_timeout</>.
+        </para>
+       </listitem>
+      </varlistentry>
  
       </variablelist>
      </sect2>
***************
*** 2397,2402 **** SET ENABLE_SEQSCAN TO OFF;
--- 2421,2450 ----
        </para>
        </listitem>
       </varlistentry>
+ 	 
+ 	 <varlistentry id="guc-wal-send-status-interval" xreflabel="wal_send_status_interval">
+       <term><varname>wal_send_status_interval</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>wal_send_status_interval</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
+       <para>
+        Specifies the minimum frequency for the WAL sender
+        process on the primary to send heart-beat message to the standby.
+        This parameter's value is the maximum interval, in seconds, between heat-beat.  
+ 	   Updates are sent each time it receives response from standby, or at least as
+        often as specified by this parameter.  Setting this parameter to zero
+        disables status updates completely.  This parameter can only be set in
+        the <filename>postgresql.conf</> file or on the server command line.
+        The default value is 10 seconds.
+       </para>
+       <para>
+        When <xref linkend="guc-wal-receiver-replication-timeout"> is enabled on a receiving server,
+        <varname>wal_send_status_interval</> must be enabled, and its value
+        must be less than the value of <varname>wal_receiver_replication_timeout</>.
+       </para>
+       </listitem>
+      </varlistentry>	 
  
       <varlistentry id="guc-hot-standby-feedback" xreflabel="hot_standby">
        <term><varname>hot_standby_feedback</varname> (<type>boolean</type>)</term>
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 62,67 **** walrcv_connect_type walrcv_connect = NULL;
--- 62,69 ----
  walrcv_receive_type walrcv_receive = NULL;
  walrcv_send_type walrcv_send = NULL;
  walrcv_disconnect_type walrcv_disconnect = NULL;
+ int			wal_receiver_replication_timeout = 60 * 1000;	/* maximum time to receive one
+ 												 * WAL data message */
  
  #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
***************
*** 174,179 **** WalReceiverMain(void)
--- 176,184 ----
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
  
+ 	TimestampTz last_recv_timestamp;
+ 	TimestampTz timeout = 0;
+ 
  	/*
  	 * WalRcv should be set up already (if we are a backend, we inherit this
  	 * by fork() or EXEC_BACKEND mechanism from the postmaster).
***************
*** 282,287 **** WalReceiverMain(void)
--- 287,295 ----
  	MemSet(&reply_message, 0, sizeof(reply_message));
  	MemSet(&feedback_message, 0, sizeof(feedback_message));
  
+ 	/* Initialize the last recv timestamp */
+ 	last_recv_timestamp = GetCurrentTimestamp();
+ 
  	/* Loop until end-of-streaming or error */
  	for (;;)
  	{
***************
*** 316,327 **** WalReceiverMain(void)
--- 324,343 ----
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
+ 			/* Something is received from master, so reset last receive time*/
+ 			last_recv_timestamp = GetCurrentTimestamp();
+ 			
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
+ 			{
+ 				/* Something is received from master, so reset last receive time*/
+ 				last_recv_timestamp = GetCurrentTimestamp();
+ 				
  				XLogWalRcvProcessMsg(type, buf, len);
+ 			}
  
  			/* Let the master know that we received some data. */
  			XLogWalRcvSendReply();
***************
*** 334,339 **** WalReceiverMain(void)
--- 350,369 ----
  		}
  		else
  		{
+ 			/* Check if time since last receive from standby has reached the configured limit
+ 			 * No need to check if it is disabled by giving value as 0*/
+ 			if (wal_receiver_replication_timeout > 0)
+ 			{
+ 				timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
+ 														  wal_receiver_replication_timeout);
+ 
+ 				if (GetCurrentTimestamp() >= timeout)
+ 				{
+ 					ereport(ERROR,
+ 						(errmsg("Could not receive any message from WalSender for configured timeout period")));
+ 				}
+ 			}
+ 		
  			/*
  			 * We didn't receive anything new, but send a status update to the
  			 * master anyway, to report any progress in applying WAL.
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 82,87 **** bool		am_cascading_walsender = false;		/* Am I cascading WAL to
--- 82,89 ----
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
  int			replication_timeout = 60 * 1000;	/* maximum time to send one
  												 * WAL data message */
+ int			wal_send_status_interval = 10 * 1000; /* send replies at least this often to standby */
+ 												 	
  /*
   * State for WalSndWakeupRequest
   */
***************
*** 832,843 **** WalSndLoop(void)
  			long		sleeptime = 10000;		/* 10 s */
  			int			wakeEvents;
  
  			wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
  				WL_SOCKET_READABLE | WL_TIMEOUT;
  
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (MyWalSnd->sendKeepalive)
  			{
  				WalSndKeepalive(output_message);
  				/* Try to flush pending output to the client */
--- 834,855 ----
  			long		sleeptime = 10000;		/* 10 s */
  			int			wakeEvents;
  
+ 			/* sleeptime should be equal to wal send interval if it is greater than zero*/
+ 			if (wal_send_status_interval > 0)
+ 			{
+ 				sleeptime = wal_send_status_interval*1000;
+ 			}			
+ 
  			wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
  				WL_SOCKET_READABLE | WL_TIMEOUT;
  
+ 			/* 
+ 			  * send keepalive message if sendkeepalive is enabled or WAL send status 
+ 			  * interval is greater than zero.
+ 			  */
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (MyWalSnd->sendKeepalive || (wal_send_status_interval > 0))
  			{
  				WalSndKeepalive(output_message);
  				/* Try to flush pending output to the client */
***************
*** 850,856 **** WalSndLoop(void)
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
  													  replication_timeout);
! 				sleeptime = 1 + (replication_timeout / 10);
  			}
  
  			/* Sleep until something happens or replication timeout */
--- 862,871 ----
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
  													  replication_timeout);
! 				if (wal_send_status_interval <= 0)
! 				{
! 					sleeptime = 1 + (replication_timeout / 10);
! 				}
  			}
  
  			/* Sleep until something happens or replication timeout */
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1596,1601 **** static struct config_int ConfigureNamesInt[] =
--- 1596,1612 ----
  	},
  
  	{
+ 		{"wal_receiver_replication_timeout", PGC_SIGHUP, REPLICATION_STANDBY,
+ 			gettext_noop("Sets the maximum wait time to receive data from master."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&wal_receiver_replication_timeout,
+ 		60 * 1000, 0, INT_MAX,
+ 		NULL, NULL, NULL
+ 	},
+ 
+ 	{
  		{"max_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
  			gettext_noop("Sets the maximum number of concurrent connections."),
  			NULL
***************
*** 2030,2035 **** static struct config_int ConfigureNamesInt[] =
--- 2041,2057 ----
  	},
  
  	{
+ 		{"wal_send_status_interval", PGC_SIGHUP, REPLICATION_SENDING,
+ 			gettext_noop("Sets the maximum interval between WAL sender reports to the standby."),
+ 			NULL,
+ 			GUC_UNIT_S
+ 		},
+ 		&wal_send_status_interval,
+ 		10, 0, INT_MAX / 1000,
+ 		NULL, NULL, NULL
+ 	},
+ 
+ 	{
  		{"commit_delay", PGC_USERSET, WAL_SETTINGS,
  			gettext_noop("Sets the delay in microseconds between transaction commit and "
  						 "flushing WAL to disk."),
***************
*** 2381,2387 **** static struct config_int ConfigureNamesInt[] =
  		1024, 100, 102400,
  		NULL, NULL, NULL
  	},
! 
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
--- 2403,2409 ----
  		1024, 100, 102400,
  		NULL, NULL, NULL
  	},
! 		
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 220,225 ****
--- 220,227 ----
  				# comma-separated list of application_name
  				# from standby(s); '*' = all
  #vacuum_defer_cleanup_age = 0	# number of xacts by which cleanup is delayed
+ #wal_send_status_interval = 10s	# send replies at least this often to standby
+ 					#  in seconds; 0 disables					
  
  # - Standby Servers -
  
***************
*** 234,242 ****
  					# when reading streaming WAL;
  					# -1 allows indefinite delay
  #wal_receiver_status_interval = 10s	# send replies at least this often
! 					# 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
  
  
  #------------------------------------------------------------------------------
--- 236,246 ----
  					# when reading streaming WAL;
  					# -1 allows indefinite delay
  #wal_receiver_status_interval = 10s	# send replies at least this often
! 					# in seconds; 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
+ #wal_receiver_replication_timeout = 60s	# in milliseconds; 0 disables; time 
+ 					# till receiver waits for communication from master.
  
  
  #------------------------------------------------------------------------------
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 19,24 ****
--- 19,25 ----
  
  extern int	wal_receiver_status_interval;
  extern bool hot_standby_feedback;
+ extern int wal_receiver_replication_timeout;
  
  /*
   * MAXCONNINFO: maximum size of a connection string.
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 26,31 **** extern bool wake_wal_senders;
--- 26,33 ----
  /* user-settable parameters */
  extern int	max_wal_senders;
  extern int	replication_timeout;
+ extern int	wal_send_status_interval;
+ 
  
  extern void WalSenderMain(void) __attribute__((noreturn));
  extern void WalSndSignals(void);

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 2152,2161 **** SET ENABLE_SEQSCAN TO OFF;
         </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>
--- 2152,2161 ----
         </listitem>
        </varlistentry>
  
!      <varlistentry id="guc-wal-sender-timeout" xreflabel="wal_sender_timeout">
!       <term><varname>wal_sender_timeout</varname> (<type>integer</type>)</term>
        <indexterm>
!        <primary><varname>wal_sender_timeout</> configuration parameter</primary>
        </indexterm>
        <listitem>
         <para>
***************
*** 2167,2177 **** SET ENABLE_SEQSCAN TO OFF;
          the <filename>postgresql.conf</> file or on the server command line.
          The default value is 60 seconds.
         </para>
         <para>
!         To prevent connections from being terminated prematurely,
!         <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</>.
         </para>
        </listitem>
       </varlistentry>
--- 2167,2189 ----
          the <filename>postgresql.conf</> file or on the server command line.
          The default value is 60 seconds.
         </para>
+       </listitem>
+      </varlistentry>
+ 	 
+ 	 <varlistentry id="guc-wal-receiver-timeout" xreflabel="wal_receiver_timeout">
+       <term><varname>wal_receiver_timeout</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>wal_receiver_timeout</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
         <para>
!         Terminate replication connections that are inactive longer
!         than the specified number of milliseconds. This is useful for
!         the receiving standby server to detect a primary node crash or network outage.
!         A value of zero disables the timeout mechanism.  This parameter
!         can only be set in
!         the <filename>postgresql.conf</> file or on the server command line.
!         The default value is 60 seconds.
         </para>
        </listitem>
       </varlistentry>
***************
*** 2390,2400 **** SET ENABLE_SEQSCAN TO OFF;
         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 a sending server,
-        <varname>wal_receiver_status_interval</> must be enabled, and its value
-        must be less than the value of <varname>replication_timeout</>.
-       </para>
        </listitem>
       </varlistentry>
  
--- 2402,2407 ----
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 38,43 ****
--- 38,44 ----
  #include <signal.h>
  #include <unistd.h>
  
+ #include "access/transam.h"
  #include "access/xlog_internal.h"
  #include "libpq/pqsignal.h"
  #include "miscadmin.h"
***************
*** 62,67 **** walrcv_connect_type walrcv_connect = NULL;
--- 63,70 ----
  walrcv_receive_type walrcv_receive = NULL;
  walrcv_send_type walrcv_send = NULL;
  walrcv_disconnect_type walrcv_disconnect = NULL;
+ int			wal_receiver_timeout = 60 * 1000;	/* maximum time to receive one
+ 												 * WAL data message */
  
  #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
***************
*** 121,127 **** static void WalRcvDie(int code, Datum arg);
  static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
  static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
  static void XLogWalRcvFlush(bool dying);
! static void XLogWalRcvSendReply(void);
  static void XLogWalRcvSendHSFeedback(void);
  static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
  
--- 124,130 ----
  static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
  static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
  static void XLogWalRcvFlush(bool dying);
! static void XLogWalRcvSendReply(bool sendImmediate);
  static void XLogWalRcvSendHSFeedback(void);
  static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
  
***************
*** 174,179 **** WalReceiverMain(void)
--- 177,185 ----
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
  
+ 	TimestampTz last_recv_timestamp;
+ 	TimestampTz timeout = 0;
+ 
  	/*
  	 * WalRcv should be set up already (if we are a backend, we inherit this
  	 * by fork() or EXEC_BACKEND mechanism from the postmaster).
***************
*** 282,287 **** WalReceiverMain(void)
--- 288,296 ----
  	MemSet(&reply_message, 0, sizeof(reply_message));
  	MemSet(&feedback_message, 0, sizeof(feedback_message));
  
+ 	/* Initialize the last recv timestamp */
+ 	last_recv_timestamp = GetCurrentTimestamp();
+ 
  	/* Loop until end-of-streaming or error */
  	for (;;)
  	{
***************
*** 316,330 **** WalReceiverMain(void)
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
  				XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Let the master know that we received some data. */
! 			XLogWalRcvSendReply();
  
  			/*
  			 * If we've written some records, flush them to disk and let the
--- 325,347 ----
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
+ 			/* Something is received from master, so reset last receive time*/
+ 			last_recv_timestamp = GetCurrentTimestamp();
+ 			
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
+ 			{
+ 				/* Something is received from master, so reset last receive time*/
+ 				last_recv_timestamp = GetCurrentTimestamp();
+ 				
  				XLogWalRcvProcessMsg(type, buf, len);
+ 			}
  
  			/* Let the master know that we received some data. */
! 			XLogWalRcvSendReply(false);
  
  			/*
  			 * If we've written some records, flush them to disk and let the
***************
*** 334,345 **** WalReceiverMain(void)
  		}
  		else
  		{
! 			/*
! 			 * We didn't receive anything new, but send a status update to the
! 			 * master anyway, to report any progress in applying WAL.
! 			 */
! 			XLogWalRcvSendReply();
! 			XLogWalRcvSendHSFeedback();
  		}
  	}
  }
--- 351,380 ----
  		}
  		else
  		{
! 			/* Check if time since last receive from standby has reached the configured limit
! 			 * No need to check if it is disabled by giving value as 0*/
! 			if (wal_receiver_timeout > 0)
! 			{
! 				timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
! 														  wal_receiver_timeout);
! 
! 				if (GetCurrentTimestamp() >= timeout)
! 				{
! 					ereport(ERROR,
! 						(errmsg("Could not receive any message from WalSender for configured timeout period")));
! 				}
! 
! 				/*
! 				 * We didn't receive anything new, for half of receiver replication timeout.
! 				 */
! 				timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
! 														  (wal_receiver_timeout/2));
! 
! 				if (GetCurrentTimestamp() >= timeout)
! 				{
! 					XLogWalRcvSendHSFeedback();
! 				}								
! 			}		
  		}
  	}
  }
***************
*** 460,465 **** XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
--- 495,506 ----
  				memcpy(&keepalive, buf, sizeof(PrimaryKeepaliveMessage));
  
  				ProcessWalSndrMessage(keepalive.walEnd, keepalive.sendTime);
+ 
+ 				/* For heart-beat message from primary, send an immediate reply*/
+ 				if (keepalive.msgType != 'r')
+ 				{
+ 					XLogWalRcvSendReply(true);
+ 				}				
  				break;
  			}
  		default:
***************
*** 610,636 **** XLogWalRcvFlush(bool dying)
  		/* Also let the master know that we made some progress */
  		if (!dying)
  		{
! 			XLogWalRcvSendReply();
! 			XLogWalRcvSendHSFeedback();
  		}
  	}
  }
  
  /*
!  * Send reply message to primary, indicating our current XLOG positions and
!  * the current time.
   */
  static void
! XLogWalRcvSendReply(void)
  {
  	char		buf[sizeof(StandbyReplyMessage) + 1];
  	TimestampTz now;
  
  	/*
  	 * If the user doesn't want status to be reported to the master, be sure
  	 * to exit before doing anything at all.
  	 */
! 	if (wal_receiver_status_interval <= 0)
  		return;
  
  	/* Get current timestamp. */
--- 651,684 ----
  		/* Also let the master know that we made some progress */
  		if (!dying)
  		{
! 			XLogWalRcvSendReply(false);
  		}
  	}
  }
  
  /*
!  * Send reply message to primary, indicating our current XLOG positions, oldest
!  * xmin and the current time.
!  * The parameter sendImmediate is used to decide whether the reply has to be 
!  *  send immediately or after wal_receiver_status_interval.
!  * If the reply is getting sent because of heart-beat from primary, then this param
!  * should be true otherwise false.
   */
  static void
! XLogWalRcvSendReply(bool sendImmediate)
  {
  	char		buf[sizeof(StandbyReplyMessage) + 1];
  	TimestampTz now;
+ 	TransactionId nextXid;
+ 	uint32		nextEpoch;
+ 	TransactionId xmin;
+ 	
  
  	/*
  	 * If the user doesn't want status to be reported to the master, be sure
  	 * to exit before doing anything at all.
  	 */
! 	if (!sendImmediate && wal_receiver_status_interval <= 0)
  		return;
  
  	/* Get current timestamp. */
***************
*** 645,651 **** XLogWalRcvSendReply(void)
  	 * this is only for reporting purposes and only on idle systems, that's
  	 * probably OK.
  	 */
! 	if (XLByteEQ(reply_message.write, LogstreamResult.Write)
  		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
  		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
  									   wal_receiver_status_interval * 1000))
--- 693,700 ----
  	 * this is only for reporting purposes and only on idle systems, that's
  	 * probably OK.
  	 */
! 	if (!sendImmediate
! 		&& XLByteEQ(reply_message.write, LogstreamResult.Write)
  		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
  		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
  									   wal_receiver_status_interval * 1000))
***************
*** 654,660 **** XLogWalRcvSendReply(void)
  	/* Construct a new message */
  	reply_message.write = LogstreamResult.Write;
  	reply_message.flush = LogstreamResult.Flush;
! 	reply_message.apply = GetXLogReplayRecPtr(NULL);
  	reply_message.sendTime = now;
  
  	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
--- 703,709 ----
  	/* Construct a new message */
  	reply_message.write = LogstreamResult.Write;
  	reply_message.flush = LogstreamResult.Flush;
! 	reply_message.apply = GetXLogReplayRecPtr(NULL);	
  	reply_message.sendTime = now;
  
  	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
***************
*** 662,667 **** XLogWalRcvSendReply(void)
--- 711,749 ----
  		 (uint32) (reply_message.flush >> 32), (uint32) reply_message.flush,
  		 (uint32) (reply_message.apply >> 32), (uint32) reply_message.apply);
  
+ 	if (hot_standby_feedback && HotStandbyActive())
+ 	{
+ 		/*
+ 		 * Make the expensive call to get the oldest xmin once we are certain
+ 		 * everything else has been checked.
+ 		 */
+ 		xmin = GetOldestXmin(true, false);
+ 
+ 		/*
+ 		 * Get epoch and adjust if nextXid and oldestXmin are different sides of
+ 		 * the epoch boundary.
+ 		 */
+ 		GetNextXidAndEpoch(&nextXid, &nextEpoch);
+ 		if (nextXid < xmin)
+ 			nextEpoch--;
+ 
+ 		/*
+ 		 * Always send feedback message.
+ 		 */
+ 		reply_message.xmin = xmin;
+ 		reply_message.epoch = nextEpoch;
+ 
+ 		elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u",
+ 			 reply_message.xmin,
+ 			 reply_message.epoch);	
+ 	}
+ 	else
+ 	{
+ 		/* Mark xmin invalid so that while primary node process this message it
+ 		  * can find out that xmin and epoch are not sent as part of reply*/
+ 		reply_message.xmin = InvalidTransactionId;
+ 	}
+ 
  	/* Prepend with the message type and send it. */
  	buf[0] = 'r';
  	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
***************
*** 681,734 **** XLogWalRcvSendHSFeedback(void)
  	uint32		nextEpoch;
  	TransactionId xmin;
  
- 	/*
- 	 * If the user doesn't want status to be reported to the master, be sure
- 	 * to exit before doing anything at all.
- 	 */
- 	if (wal_receiver_status_interval <= 0 || !hot_standby_feedback)
- 		return;
- 
  	/* Get current timestamp. */
  	now = GetCurrentTimestamp();
  
  	/*
- 	 * Send feedback at most once per wal_receiver_status_interval.
- 	 */
- 	if (!TimestampDifferenceExceeds(feedback_message.sendTime, now,
- 									wal_receiver_status_interval * 1000))
- 		return;
- 
- 	/*
- 	 * If Hot Standby is not yet active there is nothing to send. Check this
- 	 * after the interval has expired to reduce number of calls.
- 	 */
- 	if (!HotStandbyActive())
- 		return;
- 
- 	/*
- 	 * Make the expensive call to get the oldest xmin once we are certain
- 	 * everything else has been checked.
- 	 */
- 	xmin = GetOldestXmin(true, false);
- 
- 	/*
- 	 * Get epoch and adjust if nextXid and oldestXmin are different sides of
- 	 * the epoch boundary.
- 	 */
- 	GetNextXidAndEpoch(&nextXid, &nextEpoch);
- 	if (nextXid < xmin)
- 		nextEpoch--;
- 
- 	/*
  	 * Always send feedback message.
  	 */
  	feedback_message.sendTime = now;
- 	feedback_message.xmin = xmin;
- 	feedback_message.epoch = nextEpoch;
- 
- 	elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u",
- 		 feedback_message.xmin,
- 		 feedback_message.epoch);
  
  	/* Prepend with the message type and send it. */
  	buf[0] = 'h';
--- 763,775 ----
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 80,86 **** bool		am_cascading_walsender = false;		/* Am I cascading WAL to
  
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
! int			replication_timeout = 60 * 1000;	/* maximum time to send one
  												 * WAL data message */
  /*
   * State for WalSndWakeupRequest
--- 80,86 ----
  
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
! int			wal_sender_timeout = 60 * 1000;	/* maximum time to send one
  												 * WAL data message */
  /*
   * State for WalSndWakeupRequest
***************
*** 132,142 **** static void WalSndKill(int code, Datum arg);
  static void 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);
! static void WalSndKeepalive(char *msgbuf);
  
  
  /* Main entry point for walsender process */
--- 132,142 ----
  static void XLogSend(char *msgbuf, bool *caughtup);
  static void IdentifySystem(void);
  static void StartReplication(StartReplicationCmd *cmd);
! static void ProcessStandbyMessage(char *output_message);
  static void ProcessStandbyReplyMessage(void);
! static void ProcessStandbyTransInfo(StandbyReplyMessage reply);
! static void ProcessRepliesIfAny(char *output_message);
! static void WalSndKeepalive(char *msgbuf, bool isReply);
  
  
  /* Main entry point for walsender process */
***************
*** 507,513 **** HandleReplicationCommand(const char *cmd_string)
   * Check if the remote end has closed the connection.
   */
  static void
! ProcessRepliesIfAny(void)
  {
  	unsigned char firstchar;
  	int			r;
--- 507,513 ----
   * Check if the remote end has closed the connection.
   */
  static void
! ProcessRepliesIfAny(char *output_message)
  {
  	unsigned char firstchar;
  	int			r;
***************
*** 537,543 **** ProcessRepliesIfAny(void)
  				 * 'd' means a standby reply wrapped in a CopyData packet.
  				 */
  			case 'd':
! 				ProcessStandbyMessage();
  				received = true;
  				break;
  
--- 537,543 ----
  				 * 'd' means a standby reply wrapped in a CopyData packet.
  				 */
  			case 'd':
! 				ProcessStandbyMessage(output_message);
  				received = true;
  				break;
  
***************
*** 566,572 **** ProcessRepliesIfAny(void)
   * Process a status update message received from standby.
   */
  static void
! ProcessStandbyMessage(void)
  {
  	char		msgtype;
  
--- 566,572 ----
   * Process a status update message received from standby.
   */
  static void
! ProcessStandbyMessage(char *output_message)
  {
  	char		msgtype;
  
***************
*** 595,601 **** ProcessStandbyMessage(void)
  			break;
  
  		case 'h':
! 			ProcessStandbyHSFeedbackMessage();
  			break;
  
  		default:
--- 595,605 ----
  			break;
  
  		case 'h':
! 			/* 
! 			  * Once this message is received, we should send an immediate reply 
! 			  * to standby node in order to indicate connection is alive
! 			  */
! 			WalSndKeepalive(output_message, true);
  			break;
  
  		default:
***************
*** 607,613 **** ProcessStandbyMessage(void)
  }
  
  /*
!  * Regular reply from standby advising of WAL positions on standby server.
   */
  static void
  ProcessStandbyReplyMessage(void)
--- 611,617 ----
  }
  
  /*
!  * Regular reply from standby advising of WAL positions and transaction on standby server.
   */
  static void
  ProcessStandbyReplyMessage(void)
***************
*** 636,659 **** ProcessStandbyReplyMessage(void)
  		SpinLockRelease(&walsnd->mutex);
  	}
  
  	if (!am_cascading_walsender)
  		SyncRepReleaseWaiters();
  }
  
  /*
!  * Hot Standby feedback
   */
  static void
! ProcessStandbyHSFeedbackMessage(void)
  {
- 	StandbyHSFeedbackMessage reply;
  	TransactionId nextXid;
  	uint32		nextEpoch;
  
- 	/* Decipher the reply message */
- 	pq_copymsgbytes(&reply_message, (char *) &reply,
- 					sizeof(StandbyHSFeedbackMessage));
- 
  	elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
  		 reply.xmin,
  		 reply.epoch);
--- 640,660 ----
  		SpinLockRelease(&walsnd->mutex);
  	}
  
+ 	ProcessStandbyTransInfo(reply);
+ 
  	if (!am_cascading_walsender)
  		SyncRepReleaseWaiters();
  }
  
  /*
!  * Stores the transaction info received from standby
   */
  static void
! ProcessStandbyTransInfo(StandbyReplyMessage reply)
  {
  	TransactionId nextXid;
  	uint32		nextEpoch;
  
  	elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
  		 reply.xmin,
  		 reply.epoch);
***************
*** 765,771 **** WalSndLoop(void)
  		}
  
  		/* Check for input from the client */
! 		ProcessRepliesIfAny();
  
  		/*
  		 * If we don't have any pending data in the output buffer, try to send
--- 766,772 ----
  		}
  
  		/* Check for input from the client */
! 		ProcessRepliesIfAny(output_message);
  
  		/*
  		 * If we don't have any pending data in the output buffer, try to send
***************
*** 835,856 **** WalSndLoop(void)
  			wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
  				WL_SOCKET_READABLE | WL_TIMEOUT;
  
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (MyWalSnd->sendKeepalive)
  			{
! 				WalSndKeepalive(output_message);
  				/* Try to flush pending output to the client */
  				if (pq_flush_if_writable() != 0)
  					break;
  			}
  
  			/* Determine time until replication timeout */
! 			if (replication_timeout > 0)
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
! 													  replication_timeout);
! 				sleeptime = 1 + (replication_timeout / 10);
  			}
  
  			/* Sleep until something happens or replication timeout */
--- 836,867 ----
  			wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
  				WL_SOCKET_READABLE | WL_TIMEOUT;
  
+ 			/*
+ 			  * Check if half of wal_sender_timeout has lapsed without receiving any reply from standby
+ 			  * then send a keep-alive message to standby to detect standby status
+ 			  */
+ 			timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
+ 												  (wal_sender_timeout/2));
+ 			/* 
+ 			  * send keepalive message if sendkeepalive is enabled or WAL send status 
+ 			  * interval is greater than zero.
+ 			  */
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (MyWalSnd->sendKeepalive || (wal_sender_timeout > 0 && GetCurrentTimestamp() >= timeout))
  			{
! 				WalSndKeepalive(output_message, false);
  				/* Try to flush pending output to the client */
  				if (pq_flush_if_writable() != 0)
  					break;
  			}
  
  			/* Determine time until replication timeout */
! 			if (wal_sender_timeout > 0)
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
! 													  wal_sender_timeout);
! 				sleeptime = 1 + (wal_sender_timeout / 10);
  			}
  
  			/* Sleep until something happens or replication timeout */
***************
*** 862,868 **** WalSndLoop(void)
  			 * possibility that the client replied just as we reached the
  			 * timeout ... he's supposed to reply *before* that.
  			 */
! 			if (replication_timeout > 0 &&
  				GetCurrentTimestamp() >= timeout)
  			{
  				/*
--- 873,879 ----
  			 * possibility that the client replied just as we reached the
  			 * timeout ... he's supposed to reply *before* that.
  			 */
! 			if (wal_sender_timeout > 0 &&
  				GetCurrentTimestamp() >= timeout)
  			{
  				/*
***************
*** 1617,1630 **** pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
  	return (Datum) 0;
  }
  
  static void
! WalSndKeepalive(char *msgbuf)
  {
  	PrimaryKeepaliveMessage keepalive_message;
  
  	/* Construct a new message */
  	keepalive_message.walEnd = sentPtr;
  	keepalive_message.sendTime = GetCurrentTimestamp();
  
  	elog(DEBUG2, "sending replication keepalive");
  
--- 1628,1655 ----
  	return (Datum) 0;
  }
  
+ /* 
+   * This function is used to send keepalive message to standby.
+   * Depending on the parameter isReply value as true or false,
+   * it sends reply or heart-beat message to standby respectively
+   */
+  
  static void
! WalSndKeepalive(char *msgbuf, bool isReply)
  {
  	PrimaryKeepaliveMessage keepalive_message;
  
  	/* Construct a new message */
  	keepalive_message.walEnd = sentPtr;
  	keepalive_message.sendTime = GetCurrentTimestamp();
+ 	if (isReply)
+ 	{
+ 		keepalive_message.msgType = 'r';
+ 	}
+ 	else
+ 	{
+ 		keepalive_message.msgType = 'h';
+ 	}
  
  	elog(DEBUG2, "sending replication keepalive");
  
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1596,1601 **** static struct config_int ConfigureNamesInt[] =
--- 1596,1612 ----
  	},
  
  	{
+ 		{"wal_receiver_timeout", PGC_SIGHUP, REPLICATION_STANDBY,
+ 			gettext_noop("Sets the maximum wait time to receive data from master."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&wal_receiver_timeout,
+ 		60 * 1000, 0, INT_MAX,
+ 		NULL, NULL, NULL
+ 	},
+ 
+ 	{
  		{"max_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
  			gettext_noop("Sets the maximum number of concurrent connections."),
  			NULL
***************
*** 2019,2030 **** static struct config_int ConfigureNamesInt[] =
  	},
  
  	{
! 		{"replication_timeout", PGC_SIGHUP, REPLICATION_SENDING,
  			gettext_noop("Sets the maximum time to wait for WAL replication."),
  			NULL,
  			GUC_UNIT_MS
  		},
! 		&replication_timeout,
  		60 * 1000, 0, INT_MAX,
  		NULL, NULL, NULL
  	},
--- 2030,2041 ----
  	},
  
  	{
! 		{"wal_sender_timeout", PGC_SIGHUP, REPLICATION_SENDING,
  			gettext_noop("Sets the maximum time to wait for WAL replication."),
  			NULL,
  			GUC_UNIT_MS
  		},
! 		&wal_sender_timeout,
  		60 * 1000, 0, INT_MAX,
  		NULL, NULL, NULL
  	},
***************
*** 2381,2387 **** static struct config_int ConfigureNamesInt[] =
  		1024, 100, 102400,
  		NULL, NULL, NULL
  	},
! 
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
--- 2392,2398 ----
  		1024, 100, 102400,
  		NULL, NULL, NULL
  	},
! 		
  	/* End-of-list marker */
  	{
  		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 210,216 ****
  #max_wal_senders = 0		# max number of walsender processes
  				# (change requires restart)
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
! #replication_timeout = 60s	# in milliseconds; 0 disables
  
  # - Master Server -
  
--- 210,216 ----
  #max_wal_senders = 0		# max number of walsender processes
  				# (change requires restart)
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
! #wal_sender_timeout = 60s	# in milliseconds; 0 disables
  
  # - Master Server -
  
***************
*** 234,242 ****
  					# when reading streaming WAL;
  					# -1 allows indefinite delay
  #wal_receiver_status_interval = 10s	# send replies at least this often
! 					# 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
  
  
  #------------------------------------------------------------------------------
--- 234,244 ----
  					# when reading streaming WAL;
  					# -1 allows indefinite delay
  #wal_receiver_status_interval = 10s	# send replies at least this often
! 					# in seconds; 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
+ #wal_receiver_timeout = 60s	# in milliseconds; 0 disables; time 
+ 					# till receiver waits for communication from master.
  
  
  #------------------------------------------------------------------------------
*** a/src/include/replication/walprotocol.h
--- b/src/include/replication/walprotocol.h
***************
*** 27,32 **** typedef struct
--- 27,37 ----
  
  	/* Sender's system clock at the time of transmission */
  	TimestampTz sendTime;
+ 
+ 	/* This is the message type. If its type is r i.e. reply then it overrides the keepalive
+ 	 * and it serves the purpose of reply for heart-beat message from standby.
+ 	 */
+ 	char msgType; 
  } WalSndrMessage;
  
  
***************
*** 78,83 **** typedef struct
--- 83,102 ----
  	XLogRecPtr	flush;
  	XLogRecPtr	apply;
  
+ 	/* 
+ 	  * Transaction information (Earlier these field were used to be with HS Feedback 
+ 	  * message but now both of the messages are combined so combining structure
+ 	  * members also
+ 	  */
+ 	/*
+ 	 * The current xmin and epoch from the standby, for Hot Standby feedback.
+ 	 * This may be invalid if the standby-side does not support feedback, or
+ 	 * Hot Standby is not yet available.
+ 	 */	  
+ 	TransactionId xmin;
+ 	uint32		epoch;
+ 	  
+ 
  	/* Sender's system clock at the time of transmission */
  	TimestampTz sendTime;
  } StandbyReplyMessage;
***************
*** 90,103 **** typedef struct
   */
  typedef struct
  {
- 	/*
- 	 * The current xmin and epoch from the standby, for Hot Standby feedback.
- 	 * This may be invalid if the standby-side does not support feedback, or
- 	 * Hot Standby is not yet available.
- 	 */
- 	TransactionId xmin;
- 	uint32		epoch;
- 
  	/* Sender's system clock at the time of transmission */
  	TimestampTz sendTime;
  } StandbyHSFeedbackMessage;
--- 109,114 ----
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 19,24 ****
--- 19,25 ----
  
  extern int	wal_receiver_status_interval;
  extern bool hot_standby_feedback;
+ extern int wal_receiver_timeout;
  
  /*
   * MAXCONNINFO: maximum size of a connection string.
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 25,31 **** extern bool wake_wal_senders;
  
  /* user-settable parameters */
  extern int	max_wal_senders;
! extern int	replication_timeout;
  
  extern void WalSenderMain(void) __attribute__((noreturn));
  extern void WalSndSignals(void);
--- 25,31 ----
  
  /* user-settable parameters */
  extern int	max_wal_senders;
! extern int	wal_sender_timeout;
  
  extern void WalSenderMain(void) __attribute__((noreturn));
  extern void WalSndSignals(void);

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 2236,2245 **** include 'filename'
         </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>
--- 2236,2245 ----
         </listitem>
        </varlistentry>
  
!      <varlistentry id="guc-wal-sender-timeout" xreflabel="wal_sender_timeout">
!       <term><varname>wal_sender_timeout</varname> (<type>integer</type>)</term>
        <indexterm>
!        <primary><varname>wal_sender_timeout</> configuration parameter</primary>
        </indexterm>
        <listitem>
         <para>
***************
*** 2251,2262 **** include 'filename'
          the <filename>postgresql.conf</> file or on the server command line.
          The default value is 60 seconds.
         </para>
-        <para>
-         To prevent connections from being terminated prematurely,
-         <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</>.
-        </para>
        </listitem>
       </varlistentry>
  
--- 2251,2256 ----
***************
*** 2474,2484 **** include 'filename'
         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 a sending server,
-        <varname>wal_receiver_status_interval</> must be enabled, and its value
-        must be less than the value of <varname>replication_timeout</>.
-       </para>
        </listitem>
       </varlistentry>
  
--- 2468,2473 ----
***************
*** 2507,2512 **** include 'filename'
--- 2496,2520 ----
        </listitem>
       </varlistentry>
  
+      <varlistentry id="guc-wal-receiver-timeout" xreflabel="wal_receiver_timeout">
+       <term><varname>wal_receiver_timeout</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>wal_receiver_timeout</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
+        <para>
+         Terminate replication connections that are inactive longer
+         than the specified number of milliseconds. This is useful for
+         the receiving standby server to detect a primary node crash or network
+         outage.
+         A value of zero disables the timeout mechanism.  This parameter
+         can only be set in
+         the <filename>postgresql.conf</> file or on the server command line.
+         The default value is 60 seconds.
+        </para>
+       </listitem>
+      </varlistentry>
+ 
       </variablelist>
      </sect2>
     </sect1>
*** a/doc/src/sgml/release-9.1.sgml
--- b/doc/src/sgml/release-9.1.sgml
***************
*** 3322,3328 ****
       <listitem>
        <para>
         Add
!        <link linkend="guc-replication-timeout"><varname>replication_timeout</></link>
         setting (Fujii Masao, Heikki Linnakangas)
        </para>
  
--- 3322,3328 ----
       <listitem>
        <para>
         Add
!        <varname>replication_timeout</>
         setting (Fujii Masao, Heikki Linnakangas)
        </para>
  
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 55,60 ****
--- 55,61 ----
  
  /* GUC variables */
  int			wal_receiver_status_interval;
+ int			wal_receiver_timeout;
  bool		hot_standby_feedback;
  
  /* libpqreceiver hooks to these when loaded */
***************
*** 121,127 **** static void WalRcvDie(int code, Datum arg);
  static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
  static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
  static void XLogWalRcvFlush(bool dying);
! static void XLogWalRcvSendReply(void);
  static void XLogWalRcvSendHSFeedback(void);
  static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
  
--- 122,128 ----
  static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
  static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
  static void XLogWalRcvFlush(bool dying);
! static void XLogWalRcvSendReply(bool force, bool requestReply);
  static void XLogWalRcvSendHSFeedback(void);
  static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
  
***************
*** 170,178 **** WalReceiverMain(void)
  {
  	char		conninfo[MAXCONNINFO];
  	XLogRecPtr	startpoint;
- 
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
  
  	/*
  	 * WalRcv should be set up already (if we are a backend, we inherit this
--- 171,180 ----
  {
  	char		conninfo[MAXCONNINFO];
  	XLogRecPtr	startpoint;
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
+ 	TimestampTz last_recv_timestamp;
+ 	bool		ping_sent;
  
  	/*
  	 * WalRcv should be set up already (if we are a backend, we inherit this
***************
*** 282,287 **** WalReceiverMain(void)
--- 284,293 ----
  	MemSet(&reply_message, 0, sizeof(reply_message));
  	MemSet(&feedback_message, 0, sizeof(feedback_message));
  
+ 	/* Initialize the last recv timestamp */
+ 	last_recv_timestamp = GetCurrentTimestamp();
+ 	ping_sent = false;
+ 
  	/* Loop until end-of-streaming or error */
  	for (;;)
  	{
***************
*** 316,330 **** WalReceiverMain(void)
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
  				XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Let the master know that we received some data. */
! 			XLogWalRcvSendReply();
  
  			/*
  			 * If we've written some records, flush them to disk and let the
--- 322,344 ----
  		/* Wait a while for data to arrive */
  		if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
  		{
+ 			/* Something was received from master, so reset timeout */
+ 			last_recv_timestamp = GetCurrentTimestamp();
+ 			ping_sent = false;
+ 
  			/* Accept the received data, and process it */
  			XLogWalRcvProcessMsg(type, buf, len);
  
  			/* Receive any more data we can without sleeping */
  			while (walrcv_receive(0, &type, &buf, &len))
+ 			{
+ 				last_recv_timestamp = GetCurrentTimestamp();
+ 				ping_sent = false;
  				XLogWalRcvProcessMsg(type, buf, len);
+ 			}
  
  			/* Let the master know that we received some data. */
! 			XLogWalRcvSendReply(false, false);
  
  			/*
  			 * If we've written some records, flush them to disk and let the
***************
*** 335,344 **** WalReceiverMain(void)
  		else
  		{
  			/*
! 			 * We didn't receive anything new, but send a status update to the
! 			 * master anyway, to report any progress in applying WAL.
  			 */
! 			XLogWalRcvSendReply();
  			XLogWalRcvSendHSFeedback();
  		}
  	}
--- 349,396 ----
  		else
  		{
  			/*
! 			 * We didn't receive anything new. If we haven't heard anything
! 			 * from the server for more than wal_receiver_timeout / 2,
! 			 * ping the server. Also, if it's been longer than
! 			 * wal_receiver_status_interval since the last update we sent,
! 			 * send a status update to the master anyway, to report any
! 			 * progress in applying WAL.
! 			 */
! 			bool requestReply = false;
! 
! 			/*
! 			 * Check if time since last receive from standby has reached the
! 			 * configured limit.
  			 */
! 			if (wal_receiver_timeout > 0)
! 			{
! 				TimestampTz now = GetCurrentTimestamp();
! 				TimestampTz timeout;
! 
! 				timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
! 													  wal_receiver_timeout);
! 
! 				if (now >= timeout)
! 					ereport(ERROR,
! 							(errmsg("terminating walreceiver due to timeout")));
! 
! 				/*
! 				 * We didn't receive anything new, for half of receiver
! 				 * replication timeout. Ping the server.
! 				 */
! 				if (!ping_sent)
! 				{
! 					timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
! 														  (wal_receiver_timeout/2));
! 					if (now >= timeout)
! 					{
! 						requestReply = true;
! 						ping_sent = true;
! 					}
! 				}
! 			}
! 
! 			XLogWalRcvSendReply(requestReply, requestReply);
  			XLogWalRcvSendHSFeedback();
  		}
  	}
***************
*** 460,465 **** XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
--- 512,521 ----
  				memcpy(&keepalive, buf, sizeof(PrimaryKeepaliveMessage));
  
  				ProcessWalSndrMessage(keepalive.walEnd, keepalive.sendTime);
+ 
+ 				/* If the primary requested a reply, send one immediately */
+ 				if (keepalive.replyRequested)
+ 					XLogWalRcvSendReply(true, false);
  				break;
  			}
  		default:
***************
*** 609,627 **** XLogWalRcvFlush(bool dying)
  
  		/* Also let the master know that we made some progress */
  		if (!dying)
! 		{
! 			XLogWalRcvSendReply();
! 			XLogWalRcvSendHSFeedback();
! 		}
  	}
  }
  
  /*
!  * Send reply message to primary, indicating our current XLOG positions and
!  * the current time.
   */
  static void
! XLogWalRcvSendReply(void)
  {
  	char		buf[sizeof(StandbyReplyMessage) + 1];
  	TimestampTz now;
--- 665,688 ----
  
  		/* Also let the master know that we made some progress */
  		if (!dying)
! 			XLogWalRcvSendReply(false, false);
  	}
  }
  
  /*
!  * Send reply message to primary, indicating our current XLOG positions, oldest
!  * xmin and the current time.
!  *
!  * If 'force' is not true, the message is not sent unless enough time has
!  * passed since last status update to reach wal_receiver_status_internal (or
!  * if wal_receiver_status_interval is disabled altogether).
!  *
!  * If 'requestReply' is true, requests the server to reply immediately upon
!  * receiving this message. This is used for heartbearts, when approaching
!  * wal_receiver_timeout.
   */
  static void
! XLogWalRcvSendReply(bool force, bool requestReply)
  {
  	char		buf[sizeof(StandbyReplyMessage) + 1];
  	TimestampTz now;
***************
*** 630,636 **** XLogWalRcvSendReply(void)
  	 * If the user doesn't want status to be reported to the master, be sure
  	 * to exit before doing anything at all.
  	 */
! 	if (wal_receiver_status_interval <= 0)
  		return;
  
  	/* Get current timestamp. */
--- 691,697 ----
  	 * If the user doesn't want status to be reported to the master, be sure
  	 * to exit before doing anything at all.
  	 */
! 	if (!force && wal_receiver_status_interval <= 0)
  		return;
  
  	/* Get current timestamp. */
***************
*** 645,651 **** XLogWalRcvSendReply(void)
  	 * this is only for reporting purposes and only on idle systems, that's
  	 * probably OK.
  	 */
! 	if (XLByteEQ(reply_message.write, LogstreamResult.Write)
  		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
  		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
  									   wal_receiver_status_interval * 1000))
--- 706,713 ----
  	 * this is only for reporting purposes and only on idle systems, that's
  	 * probably OK.
  	 */
! 	if (!force
! 		&& XLByteEQ(reply_message.write, LogstreamResult.Write)
  		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
  		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
  									   wal_receiver_status_interval * 1000))
***************
*** 656,661 **** XLogWalRcvSendReply(void)
--- 718,724 ----
  	reply_message.flush = LogstreamResult.Flush;
  	reply_message.apply = GetXLogReplayRecPtr(NULL);
  	reply_message.sendTime = now;
+ 	reply_message.replyRequested = requestReply;
  
  	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
  		 (uint32) (reply_message.write >> 32), (uint32) reply_message.write,
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 82,88 **** static bool	replication_started = false; /* Started streaming yet? */
  
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
! int			replication_timeout = 60 * 1000;	/* maximum time to send one
  												 * WAL data message */
  /*
   * State for WalSndWakeupRequest
--- 82,88 ----
  
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
! int			wal_sender_timeout = 60 * 1000;	/* maximum time to send one
  												 * WAL data message */
  /*
   * State for WalSndWakeupRequest
***************
*** 103,117 **** static uint32 sendOff = 0;
   */
  static XLogRecPtr sentPtr = 0;
  
  /*
!  * Buffer for processing reply messages.
   */
! static StringInfoData reply_message;
  
  /*
   * Timestamp of the last receipt of the reply from the standby.
   */
  static TimestampTz last_reply_timestamp;
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
--- 103,122 ----
   */
  static XLogRecPtr sentPtr = 0;
  
+ /* Buffer for processing reply messages. */
+ static StringInfoData reply_message;
  /*
!  * Buffer for constructing outgoing messages
!  * (1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE bytes)
   */
! static char *output_message;
  
  /*
   * Timestamp of the last receipt of the reply from the standby.
   */
  static TimestampTz last_reply_timestamp;
+ /* Have we sent a heartbeat message asking for reply, since last reply? */
+ static bool	ping_sent = false;
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
***************
*** 126,139 **** static void WalSndLastCycleHandler(SIGNAL_ARGS);
  static void WalSndLoop(void) __attribute__((noreturn));
  static void InitWalSenderSlot(void);
  static void WalSndKill(int code, Datum arg);
! static void 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);
! static void WalSndKeepalive(char *msgbuf);
  
  
  /* Initialize walsender process before entering the main command loop */
--- 131,144 ----
  static void WalSndLoop(void) __attribute__((noreturn));
  static void InitWalSenderSlot(void);
  static void WalSndKill(int code, Datum arg);
! static void XLogSend(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);
! static void WalSndKeepalive(bool requestReply);
  
  
  /* Initialize walsender process before entering the main command loop */
***************
*** 465,471 **** ProcessRepliesIfAny(void)
--- 470,479 ----
  	 * Save the last reply timestamp if we've received at least one reply.
  	 */
  	if (received)
+ 	{
  		last_reply_timestamp = GetCurrentTimestamp();
+ 		ping_sent = false;
+ 	}
  }
  
  /*
***************
*** 527,532 **** ProcessStandbyReplyMessage(void)
--- 535,544 ----
  		 (uint32) (reply.flush >> 32), (uint32) reply.flush,
  		 (uint32) (reply.apply >> 32), (uint32) reply.apply);
  
+ 	/* Send a reply if the standby requested one. */
+ 	if (reply.replyRequested)
+ 		WalSndKeepalive(false);
+ 
  	/*
  	 * Update shared state for this WalSender process based on reply data from
  	 * standby.
***************
*** 620,626 **** ProcessStandbyHSFeedbackMessage(void)
  static void
  WalSndLoop(void)
  {
- 	char	   *output_message;
  	bool		caughtup = false;
  
  	/*
--- 632,637 ----
***************
*** 638,643 **** WalSndLoop(void)
--- 649,655 ----
  
  	/* Initialize the last reply timestamp */
  	last_reply_timestamp = GetCurrentTimestamp();
+ 	ping_sent = false;
  
  	/* Loop forever, unless we get an error */
  	for (;;)
***************
*** 672,678 **** WalSndLoop(void)
  		 * caught up.
  		 */
  		if (!pq_is_send_pending())
! 			XLogSend(output_message, &caughtup);
  		else
  			caughtup = false;
  
--- 684,690 ----
  		 * caught up.
  		 */
  		if (!pq_is_send_pending())
! 			XLogSend(&caughtup);
  		else
  			caughtup = false;
  
***************
*** 708,714 **** WalSndLoop(void)
  			if (walsender_ready_to_stop)
  			{
  				/* ... let's just be real sure we're caught up ... */
! 				XLogSend(output_message, &caughtup);
  				if (caughtup && !pq_is_send_pending())
  				{
  					/* Inform the standby that XLOG streaming is done */
--- 720,726 ----
  			if (walsender_ready_to_stop)
  			{
  				/* ... let's just be real sure we're caught up ... */
! 				XLogSend(&caughtup);
  				if (caughtup && !pq_is_send_pending())
  				{
  					/* Inform the standby that XLOG streaming is done */
***************
*** 738,760 **** WalSndLoop(void)
  
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (MyWalSnd->sendKeepalive)
  			{
! 				WalSndKeepalive(output_message);
! 				/* Try to flush pending output to the client */
! 				if (pq_flush_if_writable() != 0)
! 					break;
  			}
  
  			/* Determine time until replication timeout */
! 			if (replication_timeout > 0)
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
! 													  replication_timeout);
! 				sleeptime = 1 + (replication_timeout / 10);
  			}
  
! 			/* Sleep until something happens or replication timeout */
  			ImmediateInterruptOK = true;
  			CHECK_FOR_INTERRUPTS();
  			WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents,
--- 750,783 ----
  
  			if (pq_is_send_pending())
  				wakeEvents |= WL_SOCKET_WRITEABLE;
! 			else if (wal_sender_timeout > 0 && !ping_sent)
  			{
! 				/*
! 				 * If half of wal_sender_timeout has lapsed without receiving
! 				 * any reply from standby, send a keep-alive message to standby
! 				 * requesting an immediate reply.
! 				 */
! 				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
! 													  wal_sender_timeout / 2);
! 				if (GetCurrentTimestamp() >= timeout)
! 				{
! 					WalSndKeepalive(true);
! 					ping_sent = true;
! 					/* Try to flush pending output to the client */
! 					if (pq_flush_if_writable() != 0)
! 						break;
! 				}
  			}
  
  			/* Determine time until replication timeout */
! 			if (wal_sender_timeout > 0)
  			{
  				timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
! 													  wal_sender_timeout);
! 				sleeptime = 1 + (wal_sender_timeout / 10);
  			}
  
! 			/* Sleep until something happens or we time out */
  			ImmediateInterruptOK = true;
  			CHECK_FOR_INTERRUPTS();
  			WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents,
***************
*** 766,773 **** WalSndLoop(void)
  			 * possibility that the client replied just as we reached the
  			 * timeout ... he's supposed to reply *before* that.
  			 */
! 			if (replication_timeout > 0 &&
! 				GetCurrentTimestamp() >= timeout)
  			{
  				/*
  				 * Since typically expiration of replication timeout means
--- 789,795 ----
  			 * possibility that the client replied just as we reached the
  			 * timeout ... he's supposed to reply *before* that.
  			 */
! 			if (wal_sender_timeout > 0 && GetCurrentTimestamp() >= timeout)
  			{
  				/*
  				 * Since typically expiration of replication timeout means
***************
*** 1016,1030 **** retry:
   * but not yet sent to the client, and buffer it in the libpq output
   * buffer.
   *
-  * 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.
   */
  static void
! XLogSend(char *msgbuf, bool *caughtup)
  {
  	XLogRecPtr	SendRqstPtr;
  	XLogRecPtr	startptr;
--- 1038,1048 ----
   * but not yet sent to the client, and buffer it in the libpq output
   * buffer.
   *
   * If there is no unsent WAL remaining, *caughtup is set to true, otherwise
   * *caughtup is set to false.
   */
  static void
! XLogSend(bool *caughtup)
  {
  	XLogRecPtr	SendRqstPtr;
  	XLogRecPtr	startptr;
***************
*** 1107,1119 **** 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
--- 1125,1137 ----
  	/*
  	 * OK to read and send the slice.
  	 */
! 	output_message[0] = 'w';
  
  	/*
  	 * Read the log directly into the output buffer to avoid extra memcpy
  	 * calls.
  	 */
! 	XLogRead(output_message + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
  
  	/*
  	 * We fill the message header last so that the send timestamp is taken as
***************
*** 1123,1131 **** XLogSend(char *msgbuf, bool *caughtup)
  	msghdr.walEnd = SendRqstPtr;
  	msghdr.sendTime = GetCurrentTimestamp();
  
! 	memcpy(msgbuf + 1, &msghdr, sizeof(WalDataMessageHeader));
  
! 	pq_putmessage_noblock('d', msgbuf, 1 + sizeof(WalDataMessageHeader) + nbytes);
  
  	sentPtr = endptr;
  
--- 1141,1149 ----
  	msghdr.walEnd = SendRqstPtr;
  	msghdr.sendTime = GetCurrentTimestamp();
  
! 	memcpy(output_message + 1, &msghdr, sizeof(WalDataMessageHeader));
  
! 	pq_putmessage_noblock('d', output_message, 1 + sizeof(WalDataMessageHeader) + nbytes);
  
  	sentPtr = endptr;
  
***************
*** 1492,1512 **** pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
  	return (Datum) 0;
  }
  
  static void
! WalSndKeepalive(char *msgbuf)
  {
  	PrimaryKeepaliveMessage keepalive_message;
  
  	/* Construct a new message */
  	keepalive_message.walEnd = sentPtr;
  	keepalive_message.sendTime = GetCurrentTimestamp();
  
  	elog(DEBUG2, "sending replication keepalive");
  
  	/* Prepend with the message type and send it. */
! 	msgbuf[0] = 'k';
! 	memcpy(msgbuf + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
! 	pq_putmessage_noblock('d', msgbuf, sizeof(PrimaryKeepaliveMessage) + 1);
  }
  
  /*
--- 1510,1536 ----
  	return (Datum) 0;
  }
  
+ /*
+   * This function is used to send keepalive message to standby.
+   * If requestReply is set, sets a flag in the message requesting the standby
+   * to send a message back to us, for heartbeat purposes.
+   */
  static void
! WalSndKeepalive(bool requestReply)
  {
  	PrimaryKeepaliveMessage keepalive_message;
  
  	/* Construct a new message */
  	keepalive_message.walEnd = sentPtr;
  	keepalive_message.sendTime = GetCurrentTimestamp();
+ 	keepalive_message.replyRequested = requestReply;
  
  	elog(DEBUG2, "sending replication keepalive");
  
  	/* Prepend with the message type and send it. */
! 	output_message[0] = 'k';
! 	memcpy(output_message + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
! 	pq_putmessage_noblock('d', output_message, sizeof(PrimaryKeepaliveMessage) + 1);
  }
  
  /*
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1596,1601 **** static struct config_int ConfigureNamesInt[] =
--- 1596,1612 ----
  	},
  
  	{
+ 		{"wal_receiver_timeout", PGC_SIGHUP, REPLICATION_STANDBY,
+ 			gettext_noop("Sets the maximum wait time to receive data from master."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&wal_receiver_timeout,
+ 		60 * 1000, 0, INT_MAX,
+ 		NULL, NULL, NULL
+ 	},
+ 
+ 	{
  		{"max_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
  			gettext_noop("Sets the maximum number of concurrent connections."),
  			NULL
***************
*** 2019,2030 **** static struct config_int ConfigureNamesInt[] =
  	},
  
  	{
! 		{"replication_timeout", PGC_SIGHUP, REPLICATION_SENDING,
  			gettext_noop("Sets the maximum time to wait for WAL replication."),
  			NULL,
  			GUC_UNIT_MS
  		},
! 		&replication_timeout,
  		60 * 1000, 0, INT_MAX,
  		NULL, NULL, NULL
  	},
--- 2030,2041 ----
  	},
  
  	{
! 		{"wal_sender_timeout", PGC_SIGHUP, REPLICATION_SENDING,
  			gettext_noop("Sets the maximum time to wait for WAL replication."),
  			NULL,
  			GUC_UNIT_MS
  		},
! 		&wal_sender_timeout,
  		60 * 1000, 0, INT_MAX,
  		NULL, NULL, NULL
  	},
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 210,216 ****
  #max_wal_senders = 0		# max number of walsender processes
  				# (change requires restart)
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
! #replication_timeout = 60s	# in milliseconds; 0 disables
  
  # - Master Server -
  
--- 210,216 ----
  #max_wal_senders = 0		# max number of walsender processes
  				# (change requires restart)
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
! #wal_sender_timeout = 60s	# in milliseconds; 0 disables
  
  # - Master Server -
  
***************
*** 237,242 ****
--- 237,245 ----
  					# 0 disables
  #hot_standby_feedback = off		# send info from standby to prevent
  					# query conflicts
+ #wal_receiver_timeout = 60s		# time that receiver waits for
+ 					# communication from master
+ 					# in milliseconds; 0 disables
  
  
  #------------------------------------------------------------------------------
*** a/src/include/replication/walprotocol.h
--- b/src/include/replication/walprotocol.h
***************
*** 27,32 **** typedef struct
--- 27,38 ----
  
  	/* Sender's system clock at the time of transmission */
  	TimestampTz sendTime;
+ 
+ 	/*
+ 	 * If replyRequested is set, the client should reply immediately to this
+ 	 * message, to avoid a timeout disconnect.
+ 	 */
+ 	bool		replyRequested;
  } WalSndrMessage;
  
  
***************
*** 80,85 **** typedef struct
--- 86,97 ----
  
  	/* Sender's system clock at the time of transmission */
  	TimestampTz sendTime;
+ 
+ 	/*
+ 	 * If replyRequested is set, the server should reply immediately to this
+ 	 * message, to avoid a timeout disconnect.
+ 	 */
+ 	bool		replyRequested;
  } StandbyReplyMessage;
  
  /*
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 17,23 ****
--- 17,25 ----
  #include "storage/spin.h"
  #include "pgtime.h"
  
+ /* user-settable parameters */
  extern int	wal_receiver_status_interval;
+ extern int	wal_receiver_timeout;
  extern bool hot_standby_feedback;
  
  /*
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 24,30 **** extern bool wake_wal_senders;
  
  /* user-settable parameters */
  extern int	max_wal_senders;
! extern int	replication_timeout;
  
  extern void InitWalSender(void);
  extern void exec_replication_command(const char *query_string);
--- 24,30 ----
  
  /* user-settable parameters */
  extern int	max_wal_senders;
! extern int	wal_sender_timeout;
  
  extern void InitWalSender(void);
  extern void exec_replication_command(const char *query_string);
*** a/src/include/replication/walsender_private.h
--- b/src/include/replication/walsender_private.h
***************
*** 37,43 **** typedef struct WalSnd
  	XLogRecPtr	sentPtr;		/* WAL has been sent up to this point */
  	bool		needreload;		/* does currently-open file need to be
  								 * reloaded? */
- 	bool		sendKeepalive;	/* do we send keepalives on this connection? */
  
  	/*
  	 * The xlog locations that have been written, flushed, and applied by
--- 37,42 ----

*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 674,680 **** XLogWalRcvFlush(bool dying)
   * xmin and the current time.
   *
   * If 'force' is not set, the message is only sent if enough time has
!  * passed since last status update to reach wal_receiver_status_internal.
   * If wal_receiver_status_interval is disabled altogether and 'force' is
   * false, this is a no-op.
   *
--- 674,680 ----
   * xmin and the current time.
   *
   * If 'force' is not set, the message is only sent if enough time has
!  * passed since last status update to reach wal_receiver_status_interval.
   * If wal_receiver_status_interval is disabled altogether and 'force' is
   * false, this is a no-op.
   *

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 3d72a16..5a32517 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1366,7 +1366,8 @@ The commands accepted in walsender mode are:
       WAL data is sent as a series of CopyData messages.  (This allows
       other information to be intermixed; in particular the server can send
       an ErrorResponse message if it encounters a failure after beginning
-      to stream.)  The payload in each CopyData message follows this format:
+      to stream.)  The payload of each CopyData message from server to the
+      client contains a message of one of the following formats:
      </para>
 
      <para>
@@ -1390,34 +1391,32 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The starting point of the WAL data in this message, given in
-          XLogRecPtr format.
+          The starting point of the WAL data in this message.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The current end of WAL on the server, given in
-          XLogRecPtr format.
+          The current end of WAL on the server.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The server's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
       </para>
       </listitem>
       </varlistentry>
@@ -1445,25 +1444,12 @@ The commands accepted in walsender mode are:
        continuation records can be sent in different CopyData messages.
      </para>
      <para>
-       Note that all fields within the WAL data and the above-described header
-       will be in the sending server's native format.  Endianness, and the
-       format for the timestamp, are unpredictable unless the receiver has
-       verified that the sender's system identifier matches its own
-       <filename>pg_control</> contents.
-     </para>
-     <para>
        If the WAL sender process is terminated normally (during postmaster
        shutdown), it will send a CommandComplete message before exiting.
        This might not happen during an abnormal shutdown, of course.
      </para>
 
      <para>
-       The receiving process can send replies back to the sender at any time,
-       using one of the following message formats (also in the payload of a
-       CopyData message):
-     </para>
-
-     <para>
       <variablelist>
       <varlistentry>
       <term>
@@ -1495,12 +1481,23 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The server's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
+      </para>
+      </listitem>
+      </varlistentry>
+      <varlistentry>
+      <term>
+          Byte1
+      </term>
+      <listitem>
+      <para>
+          1 means that the client should reply to this message as soon as
+          possible, to avoid a timeout disconnect. 0 otherwise.
       </para>
       </listitem>
       </varlistentry>
@@ -1512,6 +1509,12 @@ The commands accepted in walsender mode are:
      </para>
 
      <para>
+       The receiving process can send replies back to the sender at any time,
+       using one of the following message formats (also in the payload of a
+       CopyData message):
+     </para>
+
+     <para>
       <variablelist>
       <varlistentry>
       <term>
@@ -1532,45 +1535,56 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
           The location of the last WAL byte + 1 received and written to disk
-          in the standby, in XLogRecPtr format.
+          in the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
           The location of the last WAL byte + 1 flushed to disk in
-          the standby, in XLogRecPtr format.
+          the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The location of the last WAL byte + 1 applied in the standby, in
-          XLogRecPtr format.
+          The location of the last WAL byte + 1 applied in the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
+      </term>
+      <listitem>
+      <para>
+          The client's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
+      </para>
+      </listitem>
+      </varlistentry>
+      <varlistentry>
+      <term>
+          Byte1
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          If 1, the client requests the server to reply to this message
+          immediately. This can be used to ping the server, to test if
+          the connection is still healthy.
       </para>
       </listitem>
       </varlistentry>
@@ -1602,28 +1616,29 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The client's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte4
+          Int32
       </term>
       <listitem>
       <para>
-          The standby's current xmin.
+          The standby's current xmin. This may be 0, if the standby does not
+          support feedback, or is not yet in Hot Standby state.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte4
+          Int32
       </term>
       <listitem>
       <para>
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index b1accdc..b41bbcc 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -39,9 +39,9 @@
 #include <unistd.h>
 
 #include "access/xlog_internal.h"
+#include "libpq/pqformat.h"
 #include "libpq/pqsignal.h"
 #include "miscadmin.h"
-#include "replication/walprotocol.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "storage/ipc.h"
@@ -93,8 +93,7 @@ static struct
 	XLogRecPtr	Flush;			/* last byte + 1 flushed in the standby */
 }	LogstreamResult;
 
-static StandbyReplyMessage reply_message;
-static StandbyHSFeedbackMessage feedback_message;
+static StringInfoData	reply_message;
 
 /*
  * About SIGTERM handling:
@@ -279,10 +278,9 @@ WalReceiverMain(void)
 	walrcv_connect(conninfo, startpoint);
 	DisableWalRcvImmediateExit();
 
-	/* Initialize LogstreamResult, reply_message and feedback_message */
+	/* Initialize LogstreamResult and reply_message */
 	LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
-	MemSet(&reply_message, 0, sizeof(reply_message));
-	MemSet(&feedback_message, 0, sizeof(feedback_message));
+	initStringInfo(&reply_message);
 
 	/* Initialize the last recv timestamp */
 	last_recv_timestamp = GetCurrentTimestamp();
@@ -480,41 +478,52 @@ WalRcvQuickDieHandler(SIGNAL_ARGS)
 static void
 XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
 {
+	StringInfoData tmpbuf;
+	int			hdrlen;
+	XLogRecPtr	dataStart;
+	XLogRecPtr	walEnd;
+	TimestampTz	sendTime;
+	bool		replyRequested;
+
+	initStringInfo(&tmpbuf);
+
 	switch (type)
 	{
 		case 'w':				/* WAL records */
 			{
-				WalDataMessageHeader msghdr;
-
-				if (len < sizeof(WalDataMessageHeader))
+				hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
+				if (len < hdrlen)
 					ereport(ERROR,
 							(errcode(ERRCODE_PROTOCOL_VIOLATION),
 							 errmsg_internal("invalid WAL message received from primary")));
-				/* memcpy is required here for alignment reasons */
-				memcpy(&msghdr, buf, sizeof(WalDataMessageHeader));
+				appendBinaryStringInfo(&tmpbuf, buf, hdrlen);
 
-				ProcessWalSndrMessage(msghdr.walEnd, msghdr.sendTime);
+				dataStart = pq_getmsgint64(&tmpbuf);
+				walEnd = pq_getmsgint64(&tmpbuf);
+				sendTime = IntegerTimestampToTimestampTz(pq_getmsgint64(&tmpbuf));
+				ProcessWalSndrMessage(walEnd, sendTime);
 
-				buf += sizeof(WalDataMessageHeader);
-				len -= sizeof(WalDataMessageHeader);
-				XLogWalRcvWrite(buf, len, msghdr.dataStart);
+				buf += hdrlen;
+				len -= hdrlen;
+				XLogWalRcvWrite(buf, len, dataStart);
 				break;
 			}
 		case 'k':				/* Keepalive */
 			{
-				PrimaryKeepaliveMessage keepalive;
-
-				if (len != sizeof(PrimaryKeepaliveMessage))
+				hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
+				if (len != hdrlen)
 					ereport(ERROR,
 							(errcode(ERRCODE_PROTOCOL_VIOLATION),
 							 errmsg_internal("invalid keepalive message received from primary")));
-				/* memcpy is required here for alignment reasons */
-				memcpy(&keepalive, buf, sizeof(PrimaryKeepaliveMessage));
+				appendBinaryStringInfo(&tmpbuf, buf, hdrlen);
+				walEnd = pq_getmsgint64(&tmpbuf);
+				sendTime = IntegerTimestampToTimestampTz(pq_getmsgint64(&tmpbuf));
+				replyRequested = pq_getmsgbyte(&tmpbuf);
 
-				ProcessWalSndrMessage(keepalive.walEnd, keepalive.sendTime);
+				ProcessWalSndrMessage(walEnd, sendTime);
 
 				/* If the primary requested a reply, send one immediately */
-				if (keepalive.replyRequested)
+				if (replyRequested)
 					XLogWalRcvSendReply(true, false);
 				break;
 			}
@@ -524,6 +533,8 @@ XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
 					 errmsg_internal("invalid replication message type %d",
 									 type)));
 	}
+
+	pfree(tmpbuf.data);
 }
 
 /*
@@ -685,7 +696,10 @@ XLogWalRcvFlush(bool dying)
 static void
 XLogWalRcvSendReply(bool force, bool requestReply)
 {
-	char		buf[sizeof(StandbyReplyMessage) + 1];
+	static XLogRecPtr writePtr = 0;
+	static XLogRecPtr flushPtr = 0;
+	XLogRecPtr	applyPtr;
+	static TimestampTz sendTime = 0;
 	TimestampTz now;
 
 	/*
@@ -708,28 +722,33 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	 * probably OK.
 	 */
 	if (!force
-		&& XLByteEQ(reply_message.write, LogstreamResult.Write)
-		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
-		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
+		&& XLByteEQ(writePtr, LogstreamResult.Write)
+		&& XLByteEQ(flushPtr, LogstreamResult.Flush)
+		&& !TimestampDifferenceExceeds(sendTime, now,
 									   wal_receiver_status_interval * 1000))
 		return;
 
 	/* Construct a new message */
-	reply_message.write = LogstreamResult.Write;
-	reply_message.flush = LogstreamResult.Flush;
-	reply_message.apply = GetXLogReplayRecPtr(NULL);
-	reply_message.sendTime = now;
-	reply_message.replyRequested = requestReply;
+	writePtr = LogstreamResult.Write;
+	flushPtr = LogstreamResult.Flush;
+	applyPtr = GetXLogReplayRecPtr(NULL);
+	sendTime = now;
+
+	resetStringInfo(&reply_message);
+	pq_sendbyte(&reply_message, 'r');
+	pq_sendint64(&reply_message, writePtr);
+	pq_sendint64(&reply_message, flushPtr);
+	pq_sendint64(&reply_message, applyPtr);
+	pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
+	pq_sendbyte(&reply_message, requestReply ? 1 : 0);
 
 	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
-		 (uint32) (reply_message.write >> 32), (uint32) reply_message.write,
-		 (uint32) (reply_message.flush >> 32), (uint32) reply_message.flush,
-		 (uint32) (reply_message.apply >> 32), (uint32) reply_message.apply);
+		 (uint32) (writePtr >> 32), (uint32) writePtr,
+		 (uint32) (flushPtr >> 32), (uint32) flushPtr,
+		 (uint32) (applyPtr >> 32), (uint32) applyPtr);
 
 	/* Prepend with the message type and send it. */
-	buf[0] = 'r';
-	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
-	walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
+	walrcv_send(reply_message.data, reply_message.len);
 }
 
 /*
@@ -739,11 +758,11 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 static void
 XLogWalRcvSendHSFeedback(void)
 {
-	char		buf[sizeof(StandbyHSFeedbackMessage) + 1];
 	TimestampTz now;
 	TransactionId nextXid;
 	uint32		nextEpoch;
 	TransactionId xmin;
+	static TimestampTz sendTime = 0;
 
 	/*
 	 * If the user doesn't want status to be reported to the master, be sure
@@ -758,7 +777,7 @@ XLogWalRcvSendHSFeedback(void)
 	/*
 	 * Send feedback at most once per wal_receiver_status_interval.
 	 */
-	if (!TimestampDifferenceExceeds(feedback_message.sendTime, now,
+	if (!TimestampDifferenceExceeds(sendTime, now,
 									wal_receiver_status_interval * 1000))
 		return;
 
@@ -786,22 +805,24 @@ XLogWalRcvSendHSFeedback(void)
 	/*
 	 * Always send feedback message.
 	 */
-	feedback_message.sendTime = now;
-	feedback_message.xmin = xmin;
-	feedback_message.epoch = nextEpoch;
+	sendTime = now;
 
 	elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u",
-		 feedback_message.xmin,
-		 feedback_message.epoch);
-
-	/* Prepend with the message type and send it. */
-	buf[0] = 'h';
-	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
-	walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
+		 xmin, nextEpoch);
+
+	/* Construct the the message and send it. */
+	resetStringInfo(&reply_message);
+	pq_sendbyte(&reply_message, 'h');
+	pq_sendint(&reply_message, xmin, 4);
+	pq_sendint(&reply_message, nextEpoch, 4);
+	walrcv_send(reply_message.data, reply_message.len);
 }
 
 /*
- * Keep track of important messages from primary.
+ * Update shared memory status upon receiving a message from primary.
+ *
+ * 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
+ * message, reported by primary.
  */
 static void
 ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 2af38f1..adeb461 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -48,7 +48,6 @@
 #include "nodes/replnodes.h"
 #include "replication/basebackup.h"
 #include "replication/syncrep.h"
-#include "replication/walprotocol.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -66,6 +65,16 @@
 #include "utils/timeout.h"
 #include "utils/timestamp.h"
 
+/*
+ * 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
+ * overhead per message in both walsender and walreceiver, but on the other
+ * hand sending large batches makes walsender less responsive to signals
+ * because signals are checked only between messages.  128kB (with
+ * default 8k blocks) seems like a reasonable guess for now.
+ */
+#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
 
 /* Array of WalSnds in shared memory */
 WalSndCtlData *WalSndCtl = NULL;
@@ -103,13 +112,10 @@ static uint32 sendOff = 0;
  */
 static XLogRecPtr sentPtr = 0;
 
-/* Buffer for processing reply messages. */
+/* Buffers for constructing outgoing messages and processing reply messages. */
+static StringInfoData output_message;
 static StringInfoData reply_message;
-/*
- * Buffer for constructing outgoing messages.
- * (1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE bytes)
- */
-static char *output_message;
+static StringInfoData tmpbuf;
 
 /*
  * Timestamp of the last receipt of the reply from the standby.
@@ -526,17 +532,24 @@ ProcessStandbyMessage(void)
 static void
 ProcessStandbyReplyMessage(void)
 {
-	StandbyReplyMessage reply;
+	XLogRecPtr	writePtr,
+				flushPtr,
+				applyPtr;
+	bool		replyRequested;
 
-	pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
+	writePtr = pq_getmsgint64(&reply_message);
+	flushPtr = pq_getmsgint64(&reply_message);
+	applyPtr = pq_getmsgint64(&reply_message);
+	(void) pq_getmsgint64(&reply_message);	/* sendTime; not used ATM */
+	replyRequested = pq_getmsgbyte(&reply_message);
 
 	elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X",
-		 (uint32) (reply.write >> 32), (uint32) reply.write,
-		 (uint32) (reply.flush >> 32), (uint32) reply.flush,
-		 (uint32) (reply.apply >> 32), (uint32) reply.apply);
+		 (uint32) (writePtr >> 32), (uint32) writePtr,
+		 (uint32) (flushPtr >> 32), (uint32) flushPtr,
+		 (uint32) (applyPtr >> 32), (uint32) applyPtr);
 
 	/* Send a reply if the standby requested one. */
-	if (reply.replyRequested)
+	if (replyRequested)
 		WalSndKeepalive(false);
 
 	/*
@@ -548,9 +561,9 @@ ProcessStandbyReplyMessage(void)
 		volatile WalSnd *walsnd = MyWalSnd;
 
 		SpinLockAcquire(&walsnd->mutex);
-		walsnd->write = reply.write;
-		walsnd->flush = reply.flush;
-		walsnd->apply = reply.apply;
+		walsnd->write = writePtr;
+		walsnd->flush = flushPtr;
+		walsnd->apply = applyPtr;
 		SpinLockRelease(&walsnd->mutex);
 	}
 
@@ -564,20 +577,21 @@ ProcessStandbyReplyMessage(void)
 static void
 ProcessStandbyHSFeedbackMessage(void)
 {
-	StandbyHSFeedbackMessage reply;
 	TransactionId nextXid;
 	uint32		nextEpoch;
+	TransactionId feedbackXmin;
+	uint32		feedbackEpoch;
 
 	/* Decipher the reply message */
-	pq_copymsgbytes(&reply_message, (char *) &reply,
-					sizeof(StandbyHSFeedbackMessage));
+	feedbackXmin = pq_getmsgint(&reply_message, 4);
+	feedbackEpoch = pq_getmsgint(&reply_message, 4);
 
 	elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
-		 reply.xmin,
-		 reply.epoch);
+		 feedbackXmin,
+		 feedbackEpoch);
 
 	/* Ignore invalid xmin (can't actually happen with current walreceiver) */
-	if (!TransactionIdIsNormal(reply.xmin))
+	if (!TransactionIdIsNormal(feedbackXmin))
 		return;
 
 	/*
@@ -589,18 +603,18 @@ ProcessStandbyHSFeedbackMessage(void)
 	 */
 	GetNextXidAndEpoch(&nextXid, &nextEpoch);
 
-	if (reply.xmin <= nextXid)
+	if (feedbackXmin <= nextXid)
 	{
-		if (reply.epoch != nextEpoch)
+		if (feedbackEpoch != nextEpoch)
 			return;
 	}
 	else
 	{
-		if (reply.epoch + 1 != nextEpoch)
+		if (feedbackEpoch + 1 != nextEpoch)
 			return;
 	}
 
-	if (!TransactionIdPrecedesOrEquals(reply.xmin, nextXid))
+	if (!TransactionIdPrecedesOrEquals(feedbackXmin, nextXid))
 		return;					/* epoch OK, but it's wrapped around */
 
 	/*
@@ -610,9 +624,9 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * cleanup conflicts on the standby server.
 	 *
 	 * There is a small window for a race condition here: although we just
-	 * checked that reply.xmin precedes nextXid, the nextXid could have gotten
+	 * checked that feedbackXmin precedes nextXid, the nextXid could have gotten
 	 * advanced between our fetching it and applying the xmin below, perhaps
-	 * far enough to make reply.xmin wrap around.  In that case the xmin we
+	 * far enough to make feedbackXmin wrap around.  In that case the xmin we
 	 * set here would be "in the future" and have no effect.  No point in
 	 * worrying about this since it's too late to save the desired data
 	 * anyway.	Assuming that the standby sends us an increasing sequence of
@@ -625,7 +639,7 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * safe, and if we're moving it backwards, well, the data is at risk
 	 * already since a VACUUM could have just finished calling GetOldestXmin.)
 	 */
-	MyPgXact->xmin = reply.xmin;
+	MyPgXact->xmin = feedbackXmin;
 }
 
 /* Main loop of walsender process that streams the WAL over Copy messages. */
@@ -635,17 +649,12 @@ WalSndLoop(void)
 	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
-	 * above, do this just once to reduce palloc overhead.
+	 * Allocate buffers that will be used for each outgoing and incoming
+	 * message.  We do this just once to reduce palloc overhead.
 	 */
+	initStringInfo(&output_message);
 	initStringInfo(&reply_message);
+	initStringInfo(&tmpbuf);
 
 	/* Initialize the last reply timestamp */
 	last_reply_timestamp = GetCurrentTimestamp();
@@ -1048,7 +1057,6 @@ XLogSend(bool *caughtup)
 	XLogRecPtr	startptr;
 	XLogRecPtr	endptr;
 	Size		nbytes;
-	WalDataMessageHeader msghdr;
 
 	/*
 	 * Attempt to send all data that's already been written out and fsync'd to
@@ -1125,25 +1133,32 @@ XLogSend(bool *caughtup)
 	/*
 	 * OK to read and send the slice.
 	 */
-	output_message[0] = 'w';
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 'w');
+
+	pq_sendint64(&output_message, startptr);		/* dataStart */
+	pq_sendint64(&output_message, SendRqstPtr);	/* walEnd */
+	pq_sendint64(&output_message, 0);			/* sendtime, filled in later */
 
 	/*
 	 * Read the log directly into the output buffer to avoid extra memcpy
 	 * calls.
 	 */
-	XLogRead(output_message + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
+	enlargeStringInfo(&output_message, nbytes);
+	XLogRead(&output_message.data[output_message.len], startptr, nbytes);
+	output_message.len += nbytes;
+	output_message.data[output_message.len] = '\0';
 
 	/*
-	 * We fill the message header last so that the send timestamp is taken as
-	 * late as possible.
+	 * Fill the send timestamp last, so that it is taken as late as possible.
 	 */
-	msghdr.dataStart = startptr;
-	msghdr.walEnd = SendRqstPtr;
-	msghdr.sendTime = GetCurrentTimestamp();
-
-	memcpy(output_message + 1, &msghdr, sizeof(WalDataMessageHeader));
+	resetStringInfo(&tmpbuf);
+	pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp());
+	Assert(tmpbuf.len == sizeof(int64));
+	memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)],
+		   &tmpbuf.data, sizeof(int64));
 
-	pq_putmessage_noblock('d', output_message, 1 + sizeof(WalDataMessageHeader) + nbytes);
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
 
 	sentPtr = endptr;
 
@@ -1518,19 +1533,17 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 static void
 WalSndKeepalive(bool requestReply)
 {
-	PrimaryKeepaliveMessage keepalive_message;
-
-	/* Construct a new message */
-	keepalive_message.walEnd = sentPtr;
-	keepalive_message.sendTime = GetCurrentTimestamp();
-	keepalive_message.replyRequested = requestReply;
-
 	elog(DEBUG2, "sending replication keepalive");
 
-	/* Prepend with the message type and send it. */
-	output_message[0] = 'k';
-	memcpy(output_message + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
-	pq_putmessage_noblock('d', output_message, sizeof(PrimaryKeepaliveMessage) + 1);
+	/* construct the message... */
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 'k');
+	pq_sendint64(&output_message, sentPtr);		/* walEnd */
+	pq_sendint64(&output_message, GetCurrentIntegerTimestamp()); /* sendTime */
+	pq_sendbyte(&output_message, requestReply ? 1 : 0);	/* replyRequested */
+
+	/* ... and send it wrapped in CopyData */
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
 }
 
 /*
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index 50ef897..823bddf 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -1286,6 +1286,50 @@ GetCurrentTimestamp(void)
 }
 
 /*
+ * GetCurrentIntegerTimestamp -- get the current operating system time as int64
+ *
+ * Result is the number of milliseconds since the Postgres epoch. If compiled
+ * with --enable-integer-datetimes, this is identical to GetCurrentTimestamp(),
+ * and is implemented as a macro.
+ */
+#ifndef HAVE_INT64_TIMESTAMP
+int64
+GetCurrentIntegerTimestamp(void)
+{
+	int64 result;
+	struct timeval tp;
+
+	gettimeofday(&tp, NULL);
+
+	result = (int64) tp.tv_sec -
+		((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
+
+	result = (result * USECS_PER_SEC) + tp.tv_usec;
+
+	return result;
+}
+#endif
+
+/*
+ * GetCurrentIntegerTimestamp -- convert an integer timestamp to native format
+ *
+ * When compiled with --enable-integer-datetimes, this is implemented as a
+ * no-op macro.
+ */
+#ifndef HAVE_INT64_TIMESTAMP
+TimestampTz
+IntegerTimestampToTimestampTz(int64 timestamp)
+{
+	TimestampTz result;
+
+	result = timestamp / USECS_PER_SEC;
+	result += (timestamp % USECS_PER_SEC) / 1000000.0;
+
+	return result;
+}
+#endif
+
+/*
  * TimestampDifference -- convert the difference between two timestamps
  *		into integer seconds and microseconds
  *
diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index 404ff91..041e97b 100644
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -21,7 +21,6 @@
 #include "postgres.h"
 #include "libpq-fe.h"
 #include "access/xlog_internal.h"
-#include "replication/walprotocol.h"
 #include "utils/datetime.h"
 #include "utils/timestamp.h"
 
@@ -33,14 +32,13 @@
 #include <sys/types.h>
 #include <unistd.h>
 
-
-/* Size of the streaming replication protocol headers */
-#define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader))
-#define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage))
-
 /* fd for currently open WAL file */
 static int	walfile = -1;
 
+static void sendint64(int64 i, char *buf);
+static int64 recvint64(char *buf);
+static bool sendFeedback(XLogRecPtr blockpos, int64 now);
+
 
 /*
  * Open a new WAL file in the specified directory. Store the name
@@ -191,35 +189,32 @@ close_walfile(char *basedir, char *walname, bool segment_complete)
  * Local version of GetCurrentTimestamp(), since we are not linked with
  * backend code.
  */
-static TimestampTz
+static int64
 localGetCurrentTimestamp(void)
 {
-	TimestampTz result;
+	int64 result;
 	struct timeval tp;
 
 	gettimeofday(&tp, NULL);
 
-	result = (TimestampTz) tp.tv_sec -
+	result = (int64) tp.tv_sec -
 		((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
 
-#ifdef HAVE_INT64_TIMESTAMP
 	result = (result * USECS_PER_SEC) + tp.tv_usec;
-#else
-	result = result + (tp.tv_usec / 1000000.0);
-#endif
 
 	return result;
 }
 
 /*
  * Local version of TimestampDifference(), since we are not
- * linked with backend code.
+ * linked with backend code. We always use integer timestamps, regardless
+ * of server setting.
  */
 static void
-localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
+localTimestampDifference(int64 start_time, int64 stop_time,
 						 long *secs, int *microsecs)
 {
-	TimestampTz diff = stop_time - start_time;
+	int64 diff = stop_time - start_time;
 
 	if (diff <= 0)
 	{
@@ -228,13 +223,8 @@ localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
 	}
 	else
 	{
-#ifdef HAVE_INT64_TIMESTAMP
 		*secs = (long) (diff / USECS_PER_SEC);
 		*microsecs = (int) (diff % USECS_PER_SEC);
-#else
-		*secs = (long) diff;
-		*microsecs = (int) ((diff - *secs) * 1000000.0);
-#endif
 	}
 }
 
@@ -243,17 +233,13 @@ localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
  * linked with backend code.
  */
 static bool
-localTimestampDifferenceExceeds(TimestampTz start_time,
-								TimestampTz stop_time,
+localTimestampDifferenceExceeds(int64 start_time,
+								int64 stop_time,
 								int msec)
 {
-	TimestampTz diff = stop_time - start_time;
+	int64 diff = stop_time - start_time;
 
-#ifdef HAVE_INT64_TIMESTAMP
 	return (diff >= msec * INT64CONST(1000));
-#else
-	return (diff * 1000.0 >= msec);
-#endif
 }
 
 /*
@@ -382,24 +368,8 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 											standby_message_timeout))
 		{
 			/* Time to send feedback! */
-			char		replybuf[sizeof(StandbyReplyMessage) + 1];
-			StandbyReplyMessage *replymsg;
-
-			replymsg = (StandbyReplyMessage *) (replybuf + 1);
-			replymsg->write = blockpos;
-			replymsg->flush = InvalidXLogRecPtr;
-			replymsg->apply = InvalidXLogRecPtr;
-			replymsg->sendTime = now;
-			replybuf[0] = 'r';
-
-			if (PQputCopyData(conn, replybuf, sizeof(replybuf)) <= 0 ||
-				PQflush(conn))
-			{
-				fprintf(stderr, _("%s: could not send feedback packet: %s"),
-						progname, PQerrorMessage(conn));
+			if (!sendFeedback(blockpos, now))
 				goto error;
-			}
-
 			last_status = now;
 		}
 
@@ -419,12 +389,11 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 			FD_SET(PQsocket(conn), &input_mask);
 			if (standby_message_timeout)
 			{
-				TimestampTz targettime;
+				int64		targettime;
 				long		secs;
 				int			usecs;
 
-				targettime = TimestampTzPlusMilliseconds(last_status,
-												standby_message_timeout - 1);
+				targettime = last_status + (standby_message_timeout - 1) * ((int64) 1000);
 				localTimestampDifference(now,
 										 targettime,
 										 &secs,
@@ -474,19 +443,13 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 					progname, PQerrorMessage(conn));
 			goto error;
 		}
+
 		if (copybuf[0] == 'k')
 		{
 			/*
 			 * keepalive message, sent in 9.2 and newer. We just ignore this
 			 * message completely, but need to skip past it in the stream.
 			 */
-			if (r != STREAMING_KEEPALIVE_SIZE)
-			{
-				fprintf(stderr,
-						_("%s: keepalive message has incorrect size %d\n"),
-						progname, r);
-				goto error;
-			}
 			continue;
 		}
 		else if (copybuf[0] != 'w')
@@ -495,15 +458,19 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 					progname, copybuf[0]);
 			goto error;
 		}
+
+/* Size of the streaming replication protocol headers */
+#define STREAMING_HEADER_SIZE (1 /* msgtype */ + 8 /* dataStart */ + 8 /* walEnd*/ + 8 /* sendTime */)
+
 		if (r < STREAMING_HEADER_SIZE + 1)
 		{
 			fprintf(stderr, _("%s: streaming header too small: %d\n"),
 					progname, r);
 			goto error;
 		}
+		blockpos = recvint64(&copybuf[1]);
 
 		/* Extract WAL location for this block */
-		memcpy(&blockpos, copybuf + 1, 8);
 		xlogoff = blockpos % XLOG_SEG_SIZE;
 
 		/*
@@ -641,3 +608,73 @@ error:
 	walfile = -1;
 	return false;
 }
+
+/*
+ * Converts an int64 to network byte order.
+ */
+static void
+sendint64(int64 i, char *buf)
+{
+	uint32		n32;
+
+	/* High order half first, since we're doing MSB-first */
+	n32 = (uint32) (i >> 32);
+	n32 = htonl(n32);
+	memcpy(&buf[0], &n32, 4);
+
+	/* Now the low order half */
+	n32 = (uint32) i;
+	n32 = htonl(n32);
+	memcpy(&buf[4], &n32, 4);
+}
+
+/*
+ * Converts an int64 from network byte order to native format.
+ */
+static int64
+recvint64(char *buf)
+{
+	int64		result;
+	uint32		h32;
+	uint32		l32;
+
+	memcpy(&h32, buf, 4);
+	memcpy(&l32, buf + 4, 4);
+	h32 = ntohl(h32);
+	l32 = ntohl(l32);
+
+	result = h32;
+	result <<= 32;
+	result |= l32;
+
+	return result;
+}
+
+/*
+ * Send a Standby Status Update message to server.
+ */
+static bool
+sendFeedback(XLogRecPtr blockpos, int64 now)
+{
+	char		replybuf[100];
+	int 		len = 0;
+
+	replybuf[len++] = 'r';
+	sendint64(blockpos, &replybuf[len]);			/* write */
+	len += 8;
+	sendint64(InvalidXLogRecPtr, &replybuf[len]);	/* flush */
+	len += 8;
+	sendint64(InvalidXLogRecPtr, &replybuf[len]);	/* apply */
+	len += 8;
+	sendint64(now, &replybuf[13]);					/* sendTime */
+	len += 8;
+
+	if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
+	{
+		fprintf(stderr, _("%s: could not send feedback packet: %s"),
+				progname, PQerrorMessage(conn));
+		return false;
+	}
+
+	return true;
+}
diff --git a/src/include/replication/walprotocol.h b/src/include/replication/walprotocol.h
deleted file mode 100644
index 396d006..0000000
--- a/src/include/replication/walprotocol.h
+++ /dev/null
@@ -1,128 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * walprotocol.h
- *	  Definitions relevant to the streaming WAL transmission protocol.
- *
- * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group
- *
- * src/include/replication/walprotocol.h
- *
- *-------------------------------------------------------------------------
- */
-#ifndef _WALPROTOCOL_H
-#define _WALPROTOCOL_H
-
-#include "access/xlogdefs.h"
-#include "datatype/timestamp.h"
-
-
-/*
- * All messages from WalSender must contain these fields to allow us to
- * correctly calculate the replication delay.
- */
-typedef struct
-{
-	/* Current end of WAL on the sender */
-	XLogRecPtr	walEnd;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-
-	/*
-	 * If replyRequested is set, the client should reply immediately to this
-	 * message, to avoid a timeout disconnect.
-	 */
-	bool		replyRequested;
-} WalSndrMessage;
-
-
-/*
- * Header for a WAL data message (message type 'w').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * The header is followed by actual WAL data.  Note that the data length is
- * not specified in the header --- it's just whatever remains in the message.
- *
- * walEnd and sendTime are not essential data, but are provided in case
- * the receiver wants to adjust its behavior depending on how far behind
- * it is.
- */
-typedef struct
-{
-	/* WAL start location of the data included in this message */
-	XLogRecPtr	dataStart;
-
-	/* Current end of WAL on the sender */
-	XLogRecPtr	walEnd;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-} WalDataMessageHeader;
-
-/*
- * Keepalive message from primary (message type 'k'). (lowercase k)
- * This is wrapped within a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef WalSndrMessage PrimaryKeepaliveMessage;
-
-/*
- * Reply message from standby (message type 'r').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef struct
-{
-	/*
-	 * The xlog locations that have been written, flushed, and applied by
-	 * standby-side. These may be invalid if the standby-side is unable to or
-	 * chooses not to report these.
-	 */
-	XLogRecPtr	write;
-	XLogRecPtr	flush;
-	XLogRecPtr	apply;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-
-	/*
-	 * If replyRequested is set, the server should reply immediately to this
-	 * message, to avoid a timeout disconnect.
-	 */
-	bool		replyRequested;
-} StandbyReplyMessage;
-
-/*
- * Hot Standby feedback from standby (message type 'h').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef struct
-{
-	/*
-	 * The current xmin and epoch from the standby, for Hot Standby feedback.
-	 * This may be invalid if the standby-side does not support feedback, or
-	 * Hot Standby is not yet available.
-	 */
-	TransactionId xmin;
-	uint32		epoch;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-} StandbyHSFeedbackMessage;
-
-/*
- * 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
- * overhead per message in both walsender and walreceiver, but on the other
- * hand sending large batches makes walsender less responsive to signals
- * because signals are checked only between messages.  128kB (with
- * default 8k blocks) seems like a reasonable guess for now.
- */
-#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
-
-#endif   /* _WALPROTOCOL_H */
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index e7cdb41..e4ca8c1 100644
--- a/src/include/utils/timestamp.h
+++ b/src/include/utils/timestamp.h
@@ -206,6 +206,13 @@ extern Datum generate_series_timestamptz(PG_FUNCTION_ARGS);
 /* Internal routines (not fmgr-callable) */
 
 extern TimestampTz GetCurrentTimestamp(void);
+#ifdef HAVE_INT64_TIMESTAMP
+#define GetCurrentIntegerTimestamp()	GetCurrentTimestamp()
+#define IntegerTimestampToTimestampTz(timestamp) (timestamp)
+#else
+extern int64 GetCurrentIntegerTimestamp(void);
+extern TimestampTz IntegerTimestampToTimestampTz(int64 timestamp);
+#endif
 
 extern void TimestampDifference(TimestampTz start_time, TimestampTz stop_time,
 					long *secs, int *microsecs);

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 3d72a16..5a32517 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1366,7 +1366,8 @@ The commands accepted in walsender mode are:
       WAL data is sent as a series of CopyData messages.  (This allows
       other information to be intermixed; in particular the server can send
       an ErrorResponse message if it encounters a failure after beginning
-      to stream.)  The payload in each CopyData message follows this format:
+      to stream.)  The payload of each CopyData message from server to the
+      client contains a message of one of the following formats:
      </para>
 
      <para>
@@ -1390,34 +1391,32 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The starting point of the WAL data in this message, given in
-          XLogRecPtr format.
+          The starting point of the WAL data in this message.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The current end of WAL on the server, given in
-          XLogRecPtr format.
+          The current end of WAL on the server.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The server's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
       </para>
       </listitem>
       </varlistentry>
@@ -1445,25 +1444,12 @@ The commands accepted in walsender mode are:
        continuation records can be sent in different CopyData messages.
      </para>
      <para>
-       Note that all fields within the WAL data and the above-described header
-       will be in the sending server's native format.  Endianness, and the
-       format for the timestamp, are unpredictable unless the receiver has
-       verified that the sender's system identifier matches its own
-       <filename>pg_control</> contents.
-     </para>
-     <para>
        If the WAL sender process is terminated normally (during postmaster
        shutdown), it will send a CommandComplete message before exiting.
        This might not happen during an abnormal shutdown, of course.
      </para>
 
      <para>
-       The receiving process can send replies back to the sender at any time,
-       using one of the following message formats (also in the payload of a
-       CopyData message):
-     </para>
-
-     <para>
       <variablelist>
       <varlistentry>
       <term>
@@ -1495,12 +1481,23 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The server's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
+      </para>
+      </listitem>
+      </varlistentry>
+      <varlistentry>
+      <term>
+          Byte1
+      </term>
+      <listitem>
+      <para>
+          1 means that the client should reply to this message as soon as
+          possible, to avoid a timeout disconnect. 0 otherwise.
       </para>
       </listitem>
       </varlistentry>
@@ -1512,6 +1509,12 @@ The commands accepted in walsender mode are:
      </para>
 
      <para>
+       The receiving process can send replies back to the sender at any time,
+       using one of the following message formats (also in the payload of a
+       CopyData message):
+     </para>
+
+     <para>
       <variablelist>
       <varlistentry>
       <term>
@@ -1532,45 +1535,56 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
           The location of the last WAL byte + 1 received and written to disk
-          in the standby, in XLogRecPtr format.
+          in the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
           The location of the last WAL byte + 1 flushed to disk in
-          the standby, in XLogRecPtr format.
+          the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The location of the last WAL byte + 1 applied in the standby, in
-          XLogRecPtr format.
+          The location of the last WAL byte + 1 applied in the standby.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
+      </term>
+      <listitem>
+      <para>
+          The client's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
+      </para>
+      </listitem>
+      </varlistentry>
+      <varlistentry>
+      <term>
+          Byte1
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          If 1, the client requests the server to reply to this message
+          immediately. This can be used to ping the server, to test if
+          the connection is still healthy.
       </para>
       </listitem>
       </varlistentry>
@@ -1602,28 +1616,29 @@ The commands accepted in walsender mode are:
       </varlistentry>
       <varlistentry>
       <term>
-          Byte8
+          Int64
       </term>
       <listitem>
       <para>
-          The server's system clock at the time of transmission,
-          given in TimestampTz format.
+          The client's system clock at the time of transmission, as
+          microseconds since midnight on 2000-01-01.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte4
+          Int32
       </term>
       <listitem>
       <para>
-          The standby's current xmin.
+          The standby's current xmin. This may be 0, if the standby does not
+          support feedback, or is not yet in Hot Standby state.
       </para>
       </listitem>
       </varlistentry>
       <varlistentry>
       <term>
-          Byte4
+          Int32
       </term>
       <listitem>
       <para>
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index b1accdc..19625c5 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -39,9 +39,9 @@
 #include <unistd.h>
 
 #include "access/xlog_internal.h"
+#include "libpq/pqformat.h"
 #include "libpq/pqsignal.h"
 #include "miscadmin.h"
-#include "replication/walprotocol.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "storage/ipc.h"
@@ -93,8 +93,8 @@ static struct
 	XLogRecPtr	Flush;			/* last byte + 1 flushed in the standby */
 }	LogstreamResult;
 
-static StandbyReplyMessage reply_message;
-static StandbyHSFeedbackMessage feedback_message;
+static StringInfoData	reply_message;
+static StringInfoData	incoming_message;
 
 /*
  * About SIGTERM handling:
@@ -279,10 +279,10 @@ WalReceiverMain(void)
 	walrcv_connect(conninfo, startpoint);
 	DisableWalRcvImmediateExit();
 
-	/* Initialize LogstreamResult, reply_message and feedback_message */
+	/* Initialize LogstreamResult and buffers for processing messages */
 	LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
-	MemSet(&reply_message, 0, sizeof(reply_message));
-	MemSet(&feedback_message, 0, sizeof(feedback_message));
+	initStringInfo(&reply_message);
+	initStringInfo(&incoming_message);
 
 	/* Initialize the last recv timestamp */
 	last_recv_timestamp = GetCurrentTimestamp();
@@ -480,41 +480,53 @@ WalRcvQuickDieHandler(SIGNAL_ARGS)
 static void
 XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
 {
+	int			hdrlen;
+	XLogRecPtr	dataStart;
+	XLogRecPtr	walEnd;
+	TimestampTz	sendTime;
+	bool		replyRequested;
+
+	resetStringInfo(&incoming_message);
+
 	switch (type)
 	{
 		case 'w':				/* WAL records */
 			{
-				WalDataMessageHeader msghdr;
-
-				if (len < sizeof(WalDataMessageHeader))
+				hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
+				if (len < hdrlen)
 					ereport(ERROR,
 							(errcode(ERRCODE_PROTOCOL_VIOLATION),
 							 errmsg_internal("invalid WAL message received from primary")));
-				/* memcpy is required here for alignment reasons */
-				memcpy(&msghdr, buf, sizeof(WalDataMessageHeader));
+				appendBinaryStringInfo(&incoming_message, buf, hdrlen);
 
-				ProcessWalSndrMessage(msghdr.walEnd, msghdr.sendTime);
+				dataStart = pq_getmsgint64(&incoming_message);
+				walEnd = pq_getmsgint64(&incoming_message);
+				sendTime = IntegerTimestampToTimestampTz(
+					pq_getmsgint64(&incoming_message));
+				ProcessWalSndrMessage(walEnd, sendTime);
 
-				buf += sizeof(WalDataMessageHeader);
-				len -= sizeof(WalDataMessageHeader);
-				XLogWalRcvWrite(buf, len, msghdr.dataStart);
+				buf += hdrlen;
+				len -= hdrlen;
+				XLogWalRcvWrite(buf, len, dataStart);
 				break;
 			}
 		case 'k':				/* Keepalive */
 			{
-				PrimaryKeepaliveMessage keepalive;
-
-				if (len != sizeof(PrimaryKeepaliveMessage))
+				hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
+				if (len != hdrlen)
 					ereport(ERROR,
 							(errcode(ERRCODE_PROTOCOL_VIOLATION),
 							 errmsg_internal("invalid keepalive message received from primary")));
-				/* memcpy is required here for alignment reasons */
-				memcpy(&keepalive, buf, sizeof(PrimaryKeepaliveMessage));
+				appendBinaryStringInfo(&incoming_message, buf, hdrlen);
+				walEnd = pq_getmsgint64(&incoming_message);
+				sendTime = IntegerTimestampToTimestampTz(
+					pq_getmsgint64(&incoming_message));
+				replyRequested = pq_getmsgbyte(&incoming_message);
 
-				ProcessWalSndrMessage(keepalive.walEnd, keepalive.sendTime);
+				ProcessWalSndrMessage(walEnd, sendTime);
 
 				/* If the primary requested a reply, send one immediately */
-				if (keepalive.replyRequested)
+				if (replyRequested)
 					XLogWalRcvSendReply(true, false);
 				break;
 			}
@@ -685,7 +697,10 @@ XLogWalRcvFlush(bool dying)
 static void
 XLogWalRcvSendReply(bool force, bool requestReply)
 {
-	char		buf[sizeof(StandbyReplyMessage) + 1];
+	static XLogRecPtr writePtr = 0;
+	static XLogRecPtr flushPtr = 0;
+	XLogRecPtr	applyPtr;
+	static TimestampTz sendTime = 0;
 	TimestampTz now;
 
 	/*
@@ -708,28 +723,33 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	 * probably OK.
 	 */
 	if (!force
-		&& XLByteEQ(reply_message.write, LogstreamResult.Write)
-		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
-		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
+		&& XLByteEQ(writePtr, LogstreamResult.Write)
+		&& XLByteEQ(flushPtr, LogstreamResult.Flush)
+		&& !TimestampDifferenceExceeds(sendTime, now,
 									   wal_receiver_status_interval * 1000))
 		return;
 
 	/* Construct a new message */
-	reply_message.write = LogstreamResult.Write;
-	reply_message.flush = LogstreamResult.Flush;
-	reply_message.apply = GetXLogReplayRecPtr(NULL);
-	reply_message.sendTime = now;
-	reply_message.replyRequested = requestReply;
+	writePtr = LogstreamResult.Write;
+	flushPtr = LogstreamResult.Flush;
+	applyPtr = GetXLogReplayRecPtr(NULL);
+	sendTime = now;
+
+	resetStringInfo(&reply_message);
+	pq_sendbyte(&reply_message, 'r');
+	pq_sendint64(&reply_message, writePtr);
+	pq_sendint64(&reply_message, flushPtr);
+	pq_sendint64(&reply_message, applyPtr);
+	pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
+	pq_sendbyte(&reply_message, requestReply ? 1 : 0);
 
 	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
-		 (uint32) (reply_message.write >> 32), (uint32) reply_message.write,
-		 (uint32) (reply_message.flush >> 32), (uint32) reply_message.flush,
-		 (uint32) (reply_message.apply >> 32), (uint32) reply_message.apply);
+		 (uint32) (writePtr >> 32), (uint32) writePtr,
+		 (uint32) (flushPtr >> 32), (uint32) flushPtr,
+		 (uint32) (applyPtr >> 32), (uint32) applyPtr);
 
 	/* Prepend with the message type and send it. */
-	buf[0] = 'r';
-	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
-	walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
+	walrcv_send(reply_message.data, reply_message.len);
 }
 
 /*
@@ -739,11 +759,11 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 static void
 XLogWalRcvSendHSFeedback(void)
 {
-	char		buf[sizeof(StandbyHSFeedbackMessage) + 1];
 	TimestampTz now;
 	TransactionId nextXid;
 	uint32		nextEpoch;
 	TransactionId xmin;
+	static TimestampTz sendTime = 0;
 
 	/*
 	 * If the user doesn't want status to be reported to the master, be sure
@@ -758,7 +778,7 @@ XLogWalRcvSendHSFeedback(void)
 	/*
 	 * Send feedback at most once per wal_receiver_status_interval.
 	 */
-	if (!TimestampDifferenceExceeds(feedback_message.sendTime, now,
+	if (!TimestampDifferenceExceeds(sendTime, now,
 									wal_receiver_status_interval * 1000))
 		return;
 
@@ -786,22 +806,25 @@ XLogWalRcvSendHSFeedback(void)
 	/*
 	 * Always send feedback message.
 	 */
-	feedback_message.sendTime = now;
-	feedback_message.xmin = xmin;
-	feedback_message.epoch = nextEpoch;
+	sendTime = now;
 
 	elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u",
-		 feedback_message.xmin,
-		 feedback_message.epoch);
-
-	/* Prepend with the message type and send it. */
-	buf[0] = 'h';
-	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
-	walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
+		 xmin, nextEpoch);
+
+	/* Construct the the message and send it. */
+	resetStringInfo(&reply_message);
+	pq_sendbyte(&reply_message, 'h');
+	pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
+	pq_sendint(&reply_message, xmin, 4);
+	pq_sendint(&reply_message, nextEpoch, 4);
+	walrcv_send(reply_message.data, reply_message.len);
 }
 
 /*
- * Keep track of important messages from primary.
+ * Update shared memory status upon receiving a message from primary.
+ *
+ * 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
+ * message, reported by primary.
  */
 static void
 ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 2af38f1..adeb461 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -48,7 +48,6 @@
 #include "nodes/replnodes.h"
 #include "replication/basebackup.h"
 #include "replication/syncrep.h"
-#include "replication/walprotocol.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -66,6 +65,16 @@
 #include "utils/timeout.h"
 #include "utils/timestamp.h"
 
+/*
+ * 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
+ * overhead per message in both walsender and walreceiver, but on the other
+ * hand sending large batches makes walsender less responsive to signals
+ * because signals are checked only between messages.  128kB (with
+ * default 8k blocks) seems like a reasonable guess for now.
+ */
+#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
 
 /* Array of WalSnds in shared memory */
 WalSndCtlData *WalSndCtl = NULL;
@@ -103,13 +112,10 @@ static uint32 sendOff = 0;
  */
 static XLogRecPtr sentPtr = 0;
 
-/* Buffer for processing reply messages. */
+/* Buffers for constructing outgoing messages and processing reply messages. */
+static StringInfoData output_message;
 static StringInfoData reply_message;
-/*
- * Buffer for constructing outgoing messages.
- * (1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE bytes)
- */
-static char *output_message;
+static StringInfoData tmpbuf;
 
 /*
  * Timestamp of the last receipt of the reply from the standby.
@@ -526,17 +532,24 @@ ProcessStandbyMessage(void)
 static void
 ProcessStandbyReplyMessage(void)
 {
-	StandbyReplyMessage reply;
+	XLogRecPtr	writePtr,
+				flushPtr,
+				applyPtr;
+	bool		replyRequested;
 
-	pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
+	writePtr = pq_getmsgint64(&reply_message);
+	flushPtr = pq_getmsgint64(&reply_message);
+	applyPtr = pq_getmsgint64(&reply_message);
+	(void) pq_getmsgint64(&reply_message);	/* sendTime; not used ATM */
+	replyRequested = pq_getmsgbyte(&reply_message);
 
 	elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X",
-		 (uint32) (reply.write >> 32), (uint32) reply.write,
-		 (uint32) (reply.flush >> 32), (uint32) reply.flush,
-		 (uint32) (reply.apply >> 32), (uint32) reply.apply);
+		 (uint32) (writePtr >> 32), (uint32) writePtr,
+		 (uint32) (flushPtr >> 32), (uint32) flushPtr,
+		 (uint32) (applyPtr >> 32), (uint32) applyPtr);
 
 	/* Send a reply if the standby requested one. */
-	if (reply.replyRequested)
+	if (replyRequested)
 		WalSndKeepalive(false);
 
 	/*
@@ -548,9 +561,9 @@ ProcessStandbyReplyMessage(void)
 		volatile WalSnd *walsnd = MyWalSnd;
 
 		SpinLockAcquire(&walsnd->mutex);
-		walsnd->write = reply.write;
-		walsnd->flush = reply.flush;
-		walsnd->apply = reply.apply;
+		walsnd->write = writePtr;
+		walsnd->flush = flushPtr;
+		walsnd->apply = applyPtr;
 		SpinLockRelease(&walsnd->mutex);
 	}
 
@@ -564,20 +577,21 @@ ProcessStandbyReplyMessage(void)
 static void
 ProcessStandbyHSFeedbackMessage(void)
 {
-	StandbyHSFeedbackMessage reply;
 	TransactionId nextXid;
 	uint32		nextEpoch;
+	TransactionId feedbackXmin;
+	uint32		feedbackEpoch;
 
 	/* Decipher the reply message */
-	pq_copymsgbytes(&reply_message, (char *) &reply,
-					sizeof(StandbyHSFeedbackMessage));
+	feedbackXmin = pq_getmsgint(&reply_message, 4);
+	feedbackEpoch = pq_getmsgint(&reply_message, 4);
 
 	elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
-		 reply.xmin,
-		 reply.epoch);
+		 feedbackXmin,
+		 feedbackEpoch);
 
 	/* Ignore invalid xmin (can't actually happen with current walreceiver) */
-	if (!TransactionIdIsNormal(reply.xmin))
+	if (!TransactionIdIsNormal(feedbackXmin))
 		return;
 
 	/*
@@ -589,18 +603,18 @@ ProcessStandbyHSFeedbackMessage(void)
 	 */
 	GetNextXidAndEpoch(&nextXid, &nextEpoch);
 
-	if (reply.xmin <= nextXid)
+	if (feedbackXmin <= nextXid)
 	{
-		if (reply.epoch != nextEpoch)
+		if (feedbackEpoch != nextEpoch)
 			return;
 	}
 	else
 	{
-		if (reply.epoch + 1 != nextEpoch)
+		if (feedbackEpoch + 1 != nextEpoch)
 			return;
 	}
 
-	if (!TransactionIdPrecedesOrEquals(reply.xmin, nextXid))
+	if (!TransactionIdPrecedesOrEquals(feedbackXmin, nextXid))
 		return;					/* epoch OK, but it's wrapped around */
 
 	/*
@@ -610,9 +624,9 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * cleanup conflicts on the standby server.
 	 *
 	 * There is a small window for a race condition here: although we just
-	 * checked that reply.xmin precedes nextXid, the nextXid could have gotten
+	 * checked that feedbackXmin precedes nextXid, the nextXid could have gotten
 	 * advanced between our fetching it and applying the xmin below, perhaps
-	 * far enough to make reply.xmin wrap around.  In that case the xmin we
+	 * far enough to make feedbackXmin wrap around.  In that case the xmin we
 	 * set here would be "in the future" and have no effect.  No point in
 	 * worrying about this since it's too late to save the desired data
 	 * anyway.	Assuming that the standby sends us an increasing sequence of
@@ -625,7 +639,7 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * safe, and if we're moving it backwards, well, the data is at risk
 	 * already since a VACUUM could have just finished calling GetOldestXmin.)
 	 */
-	MyPgXact->xmin = reply.xmin;
+	MyPgXact->xmin = feedbackXmin;
 }
 
 /* Main loop of walsender process that streams the WAL over Copy messages. */
@@ -635,17 +649,12 @@ WalSndLoop(void)
 	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
-	 * above, do this just once to reduce palloc overhead.
+	 * Allocate buffers that will be used for each outgoing and incoming
+	 * message.  We do this just once to reduce palloc overhead.
 	 */
+	initStringInfo(&output_message);
 	initStringInfo(&reply_message);
+	initStringInfo(&tmpbuf);
 
 	/* Initialize the last reply timestamp */
 	last_reply_timestamp = GetCurrentTimestamp();
@@ -1048,7 +1057,6 @@ XLogSend(bool *caughtup)
 	XLogRecPtr	startptr;
 	XLogRecPtr	endptr;
 	Size		nbytes;
-	WalDataMessageHeader msghdr;
 
 	/*
 	 * Attempt to send all data that's already been written out and fsync'd to
@@ -1125,25 +1133,32 @@ XLogSend(bool *caughtup)
 	/*
 	 * OK to read and send the slice.
 	 */
-	output_message[0] = 'w';
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 'w');
+
+	pq_sendint64(&output_message, startptr);		/* dataStart */
+	pq_sendint64(&output_message, SendRqstPtr);	/* walEnd */
+	pq_sendint64(&output_message, 0);			/* sendtime, filled in later */
 
 	/*
 	 * Read the log directly into the output buffer to avoid extra memcpy
 	 * calls.
 	 */
-	XLogRead(output_message + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
+	enlargeStringInfo(&output_message, nbytes);
+	XLogRead(&output_message.data[output_message.len], startptr, nbytes);
+	output_message.len += nbytes;
+	output_message.data[output_message.len] = '\0';
 
 	/*
-	 * We fill the message header last so that the send timestamp is taken as
-	 * late as possible.
+	 * Fill the send timestamp last, so that it is taken as late as possible.
 	 */
-	msghdr.dataStart = startptr;
-	msghdr.walEnd = SendRqstPtr;
-	msghdr.sendTime = GetCurrentTimestamp();
-
-	memcpy(output_message + 1, &msghdr, sizeof(WalDataMessageHeader));
+	resetStringInfo(&tmpbuf);
+	pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp());
+	Assert(tmpbuf.len == sizeof(int64));
+	memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)],
+		   &tmpbuf.data, sizeof(int64));
 
-	pq_putmessage_noblock('d', output_message, 1 + sizeof(WalDataMessageHeader) + nbytes);
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
 
 	sentPtr = endptr;
 
@@ -1518,19 +1533,17 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 static void
 WalSndKeepalive(bool requestReply)
 {
-	PrimaryKeepaliveMessage keepalive_message;
-
-	/* Construct a new message */
-	keepalive_message.walEnd = sentPtr;
-	keepalive_message.sendTime = GetCurrentTimestamp();
-	keepalive_message.replyRequested = requestReply;
-
 	elog(DEBUG2, "sending replication keepalive");
 
-	/* Prepend with the message type and send it. */
-	output_message[0] = 'k';
-	memcpy(output_message + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
-	pq_putmessage_noblock('d', output_message, sizeof(PrimaryKeepaliveMessage) + 1);
+	/* construct the message... */
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 'k');
+	pq_sendint64(&output_message, sentPtr);		/* walEnd */
+	pq_sendint64(&output_message, GetCurrentIntegerTimestamp()); /* sendTime */
+	pq_sendbyte(&output_message, requestReply ? 1 : 0);	/* replyRequested */
+
+	/* ... and send it wrapped in CopyData */
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
 }
 
 /*
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index 50ef897..823bddf 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -1286,6 +1286,50 @@ GetCurrentTimestamp(void)
 }
 
 /*
+ * GetCurrentIntegerTimestamp -- get the current operating system time as int64
+ *
+ * Result is the number of milliseconds since the Postgres epoch. If compiled
+ * with --enable-integer-datetimes, this is identical to GetCurrentTimestamp(),
+ * and is implemented as a macro.
+ */
+#ifndef HAVE_INT64_TIMESTAMP
+int64
+GetCurrentIntegerTimestamp(void)
+{
+	int64 result;
+	struct timeval tp;
+
+	gettimeofday(&tp, NULL);
+
+	result = (int64) tp.tv_sec -
+		((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
+
+	result = (result * USECS_PER_SEC) + tp.tv_usec;
+
+	return result;
+}
+#endif
+
+/*
+ * GetCurrentIntegerTimestamp -- convert an integer timestamp to native format
+ *
+ * When compiled with --enable-integer-datetimes, this is implemented as a
+ * no-op macro.
+ */
+#ifndef HAVE_INT64_TIMESTAMP
+TimestampTz
+IntegerTimestampToTimestampTz(int64 timestamp)
+{
+	TimestampTz result;
+
+	result = timestamp / USECS_PER_SEC;
+	result += (timestamp % USECS_PER_SEC) / 1000000.0;
+
+	return result;
+}
+#endif
+
+/*
  * TimestampDifference -- convert the difference between two timestamps
  *		into integer seconds and microseconds
  *
diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index 404ff91..97ab226 100644
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -21,7 +21,6 @@
 #include "postgres.h"
 #include "libpq-fe.h"
 #include "access/xlog_internal.h"
-#include "replication/walprotocol.h"
 #include "utils/datetime.h"
 #include "utils/timestamp.h"
 
@@ -33,14 +32,13 @@
 #include <sys/types.h>
 #include <unistd.h>
 
-
-/* Size of the streaming replication protocol headers */
-#define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader))
-#define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage))
-
 /* fd for currently open WAL file */
 static int	walfile = -1;
 
+static void sendint64(int64 i, char *buf);
+static int64 recvint64(char *buf);
+static bool sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now);
+
 
 /*
  * Open a new WAL file in the specified directory. Store the name
@@ -191,35 +189,32 @@ close_walfile(char *basedir, char *walname, bool segment_complete)
  * Local version of GetCurrentTimestamp(), since we are not linked with
  * backend code.
  */
-static TimestampTz
+static int64
 localGetCurrentTimestamp(void)
 {
-	TimestampTz result;
+	int64 result;
 	struct timeval tp;
 
 	gettimeofday(&tp, NULL);
 
-	result = (TimestampTz) tp.tv_sec -
+	result = (int64) tp.tv_sec -
 		((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
 
-#ifdef HAVE_INT64_TIMESTAMP
 	result = (result * USECS_PER_SEC) + tp.tv_usec;
-#else
-	result = result + (tp.tv_usec / 1000000.0);
-#endif
 
 	return result;
 }
 
 /*
  * Local version of TimestampDifference(), since we are not
- * linked with backend code.
+ * linked with backend code. We always use integer timestamps, regardless
+ * of server setting.
  */
 static void
-localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
+localTimestampDifference(int64 start_time, int64 stop_time,
 						 long *secs, int *microsecs)
 {
-	TimestampTz diff = stop_time - start_time;
+	int64 diff = stop_time - start_time;
 
 	if (diff <= 0)
 	{
@@ -228,13 +223,8 @@ localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
 	}
 	else
 	{
-#ifdef HAVE_INT64_TIMESTAMP
 		*secs = (long) (diff / USECS_PER_SEC);
 		*microsecs = (int) (diff % USECS_PER_SEC);
-#else
-		*secs = (long) diff;
-		*microsecs = (int) ((diff - *secs) * 1000000.0);
-#endif
 	}
 }
 
@@ -243,17 +233,13 @@ localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
  * linked with backend code.
  */
 static bool
-localTimestampDifferenceExceeds(TimestampTz start_time,
-								TimestampTz stop_time,
+localTimestampDifferenceExceeds(int64 start_time,
+								int64 stop_time,
 								int msec)
 {
-	TimestampTz diff = stop_time - start_time;
+	int64 diff = stop_time - start_time;
 
-#ifdef HAVE_INT64_TIMESTAMP
 	return (diff >= msec * INT64CONST(1000));
-#else
-	return (diff * 1000.0 >= msec);
-#endif
 }
 
 /*
@@ -382,24 +368,8 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 											standby_message_timeout))
 		{
 			/* Time to send feedback! */
-			char		replybuf[sizeof(StandbyReplyMessage) + 1];
-			StandbyReplyMessage *replymsg;
-
-			replymsg = (StandbyReplyMessage *) (replybuf + 1);
-			replymsg->write = blockpos;
-			replymsg->flush = InvalidXLogRecPtr;
-			replymsg->apply = InvalidXLogRecPtr;
-			replymsg->sendTime = now;
-			replybuf[0] = 'r';
-
-			if (PQputCopyData(conn, replybuf, sizeof(replybuf)) <= 0 ||
-				PQflush(conn))
-			{
-				fprintf(stderr, _("%s: could not send feedback packet: %s"),
-						progname, PQerrorMessage(conn));
+			if (!sendFeedback(conn, blockpos, now))
 				goto error;
-			}
-
 			last_status = now;
 		}
 
@@ -419,12 +389,11 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 			FD_SET(PQsocket(conn), &input_mask);
 			if (standby_message_timeout)
 			{
-				TimestampTz targettime;
+				int64		targettime;
 				long		secs;
 				int			usecs;
 
-				targettime = TimestampTzPlusMilliseconds(last_status,
-												standby_message_timeout - 1);
+				targettime = last_status + (standby_message_timeout - 1) * ((int64) 1000);
 				localTimestampDifference(now,
 										 targettime,
 										 &secs,
@@ -474,19 +443,13 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 					progname, PQerrorMessage(conn));
 			goto error;
 		}
+
 		if (copybuf[0] == 'k')
 		{
 			/*
 			 * keepalive message, sent in 9.2 and newer. We just ignore this
 			 * message completely, but need to skip past it in the stream.
 			 */
-			if (r != STREAMING_KEEPALIVE_SIZE)
-			{
-				fprintf(stderr,
-						_("%s: keepalive message has incorrect size %d\n"),
-						progname, r);
-				goto error;
-			}
 			continue;
 		}
 		else if (copybuf[0] != 'w')
@@ -495,15 +458,19 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 					progname, copybuf[0]);
 			goto error;
 		}
+
+/* Size of the streaming replication protocol headers */
+#define STREAMING_HEADER_SIZE (1 /* msgtype */ + 8 /* dataStart */ + 8 /* walEnd*/ + 8 /* sendTime */)
+
 		if (r < STREAMING_HEADER_SIZE + 1)
 		{
 			fprintf(stderr, _("%s: streaming header too small: %d\n"),
 					progname, r);
 			goto error;
 		}
+		blockpos = recvint64(&copybuf[1]);
 
 		/* Extract WAL location for this block */
-		memcpy(&blockpos, copybuf + 1, 8);
 		xlogoff = blockpos % XLOG_SEG_SIZE;
 
 		/*
@@ -641,3 +608,74 @@ error:
 	walfile = -1;
 	return false;
 }
+
+/*
+ * Converts an int64 to network byte order.
+ */
+static void
+sendint64(int64 i, char *buf)
+{
+	uint32		n32;
+
+	/* High order half first, since we're doing MSB-first */
+	n32 = (uint32) (i >> 32);
+	n32 = htonl(n32);
+	memcpy(&buf[0], &n32, 4);
+
+	/* Now the low order half */
+	n32 = (uint32) i;
+	n32 = htonl(n32);
+	memcpy(&buf[4], &n32, 4);
+}
+
+/*
+ * Converts an int64 from network byte order to native format.
+ */
+static int64
+recvint64(char *buf)
+{
+	int64		result;
+	uint32		h32;
+	uint32		l32;
+
+	memcpy(&h32, buf, 4);
+	memcpy(&l32, buf + 4, 4);
+	h32 = ntohl(h32);
+	l32 = ntohl(l32);
+
+	result = h32;
+	result <<= 32;
+	result |= l32;
+
+	return result;
+}
+
+/*
+ * Send a Standby Status Update message to server.
+ */
+static bool
+sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now)
+{
+	char		replybuf[1 + 8 + 8 + 8 + 8 + 1];
+	int 		len = 0;
+
+	replybuf[len++] = 'r';
+	sendint64(blockpos, &replybuf[len]);			/* write */
+	len += 8;
+	sendint64(InvalidXLogRecPtr, &replybuf[len]);	/* flush */
+	len += 8;
+	sendint64(InvalidXLogRecPtr, &replybuf[len]);	/* apply */
+	len += 8;
+	sendint64(now, &replybuf[len]);					/* sendTime */
+	len += 8;
+	replybuf[len++] = 0;							/* replyRequested */
+
+	if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
+	{
+		fprintf(stderr, _("%s: could not send feedback packet: %s"),
+				progname, PQerrorMessage(conn));
+		return false;
+	}
+
+	return true;
+}
diff --git a/src/include/replication/walprotocol.h b/src/include/replication/walprotocol.h
deleted file mode 100644
index 396d006..0000000
--- a/src/include/replication/walprotocol.h
+++ /dev/null
@@ -1,128 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * walprotocol.h
- *	  Definitions relevant to the streaming WAL transmission protocol.
- *
- * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group
- *
- * src/include/replication/walprotocol.h
- *
- *-------------------------------------------------------------------------
- */
-#ifndef _WALPROTOCOL_H
-#define _WALPROTOCOL_H
-
-#include "access/xlogdefs.h"
-#include "datatype/timestamp.h"
-
-
-/*
- * All messages from WalSender must contain these fields to allow us to
- * correctly calculate the replication delay.
- */
-typedef struct
-{
-	/* Current end of WAL on the sender */
-	XLogRecPtr	walEnd;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-
-	/*
-	 * If replyRequested is set, the client should reply immediately to this
-	 * message, to avoid a timeout disconnect.
-	 */
-	bool		replyRequested;
-} WalSndrMessage;
-
-
-/*
- * Header for a WAL data message (message type 'w').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * The header is followed by actual WAL data.  Note that the data length is
- * not specified in the header --- it's just whatever remains in the message.
- *
- * walEnd and sendTime are not essential data, but are provided in case
- * the receiver wants to adjust its behavior depending on how far behind
- * it is.
- */
-typedef struct
-{
-	/* WAL start location of the data included in this message */
-	XLogRecPtr	dataStart;
-
-	/* Current end of WAL on the sender */
-	XLogRecPtr	walEnd;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-} WalDataMessageHeader;
-
-/*
- * Keepalive message from primary (message type 'k'). (lowercase k)
- * This is wrapped within a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef WalSndrMessage PrimaryKeepaliveMessage;
-
-/*
- * Reply message from standby (message type 'r').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef struct
-{
-	/*
-	 * The xlog locations that have been written, flushed, and applied by
-	 * standby-side. These may be invalid if the standby-side is unable to or
-	 * chooses not to report these.
-	 */
-	XLogRecPtr	write;
-	XLogRecPtr	flush;
-	XLogRecPtr	apply;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-
-	/*
-	 * If replyRequested is set, the server should reply immediately to this
-	 * message, to avoid a timeout disconnect.
-	 */
-	bool		replyRequested;
-} StandbyReplyMessage;
-
-/*
- * Hot Standby feedback from standby (message type 'h').  This is wrapped within
- * a CopyData message at the FE/BE protocol level.
- *
- * Note that the data length is not specified here.
- */
-typedef struct
-{
-	/*
-	 * The current xmin and epoch from the standby, for Hot Standby feedback.
-	 * This may be invalid if the standby-side does not support feedback, or
-	 * Hot Standby is not yet available.
-	 */
-	TransactionId xmin;
-	uint32		epoch;
-
-	/* Sender's system clock at the time of transmission */
-	TimestampTz sendTime;
-} StandbyHSFeedbackMessage;
-
-/*
- * 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
- * overhead per message in both walsender and walreceiver, but on the other
- * hand sending large batches makes walsender less responsive to signals
- * because signals are checked only between messages.  128kB (with
- * default 8k blocks) seems like a reasonable guess for now.
- */
-#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
-
-#endif   /* _WALPROTOCOL_H */
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index e7cdb41..e4ca8c1 100644
--- a/src/include/utils/timestamp.h
+++ b/src/include/utils/timestamp.h
@@ -206,6 +206,13 @@ extern Datum generate_series_timestamptz(PG_FUNCTION_ARGS);
 /* Internal routines (not fmgr-callable) */
 
 extern TimestampTz GetCurrentTimestamp(void);
+#ifdef HAVE_INT64_TIMESTAMP
+#define GetCurrentIntegerTimestamp()	GetCurrentTimestamp()
+#define IntegerTimestampToTimestampTz(timestamp) (timestamp)
+#else
+extern int64 GetCurrentIntegerTimestamp(void);
+extern TimestampTz IntegerTimestampToTimestampTz(int64 timestamp);
+#endif
 
 extern void TimestampDifference(TimestampTz start_time, TimestampTz stop_time,
 					long *secs, int *microsecs);

*** a/src/bin/pg_basebackup/receivelog.c
--- b/src/bin/pg_basebackup/receivelog.c
***************
*** 256,261 **** localTimestampDifferenceExceeds(TimestampTz start_time,
--- 256,290 ----
  #endif
  }
  
+ /* Send reply to Sendert task.
+   * replyRequested is used to decide whether any immediate reply is expected 
+   * from sender
+   */
+ static bool sendReplyToSender(PGconn *conn, int64 now, bool replyRequested, XLogRecPtr blockpos)
+ {
+ 	/* Time to send feedback! */
+ 	char		replybuf[sizeof(StandbyReplyMessage) + 1];
+ 	StandbyReplyMessage *replymsg;
+ 
+ 	replymsg = (StandbyReplyMessage *) (replybuf + 1);
+ 	replymsg->write = blockpos;
+ 	replymsg->flush = InvalidXLogRecPtr;
+ 	replymsg->apply = InvalidXLogRecPtr;
+ 	replymsg->sendTime = now;
+ 	replymsg->replyRequested = replyRequested;
+ 	replybuf[0] = 'r';
+ 
+ 	if (PQputCopyData(conn, replybuf, sizeof(replybuf)) <= 0 ||
+ 		PQflush(conn))
+ 	{
+ 		fprintf(stderr, _("%s: could not send feedback packet: %s"),
+ 				progname, PQerrorMessage(conn));
+ 		return false;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /*
   * Receive a log stream starting at the specified position.
   *
***************
*** 291,296 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 320,326 ----
  	char	   *copybuf = NULL;
  	int64		last_status = -1;
  	XLogRecPtr	blockpos = InvalidXLogRecPtr;
+ 	PrimaryKeepaliveMessage keepalive;
  
  	if (sysidentifier != NULL)
  	{
***************
*** 381,402 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  			localTimestampDifferenceExceeds(last_status, now,
  											standby_message_timeout))
  		{
! 			/* Time to send feedback! */
! 			char		replybuf[sizeof(StandbyReplyMessage) + 1];
! 			StandbyReplyMessage *replymsg;
! 
! 			replymsg = (StandbyReplyMessage *) (replybuf + 1);
! 			replymsg->write = blockpos;
! 			replymsg->flush = InvalidXLogRecPtr;
! 			replymsg->apply = InvalidXLogRecPtr;
! 			replymsg->sendTime = now;
! 			replybuf[0] = 'r';
! 
! 			if (PQputCopyData(conn, replybuf, sizeof(replybuf)) <= 0 ||
! 				PQflush(conn))
  			{
- 				fprintf(stderr, _("%s: could not send feedback packet: %s"),
- 						progname, PQerrorMessage(conn));
  				goto error;
  			}
  
--- 411,419 ----
  			localTimestampDifferenceExceeds(last_status, now,
  											standby_message_timeout))
  		{
! 			/* Sending false here, because no reply is expected from server */
! 			if (sendReplyToSender(conn, now, false, blockpos) == false)
  			{
  				goto error;
  			}
  
***************
*** 487,492 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 504,527 ----
  						progname, r);
  				goto error;
  			}
+ 
+ 			/* copy the received buffer to keepalive */
+ 			memcpy(&keepalive, copybuf, sizeof(PrimaryKeepaliveMessage));
+ 
+ 			/* If as part of keepalive message from sender, an immediate reply is requested 
+ 			 * then send the same to sender.
+ 			 */
+ 			if (keepalive.replyRequested)
+ 			{
+ 				now = localGetCurrentTimestamp();
+ 				if (sendReplyToSender(conn, now, false, blockpos) == false)
+ 				{
+ 					goto error;
+ 				}
+ 
+ 				last_status = now;			
+ 			}
+ 			
  			continue;
  		}
  		else if (copybuf[0] != 'w')

*** a/src/bin/pg_basebackup/streamutil.c
--- b/src/bin/pg_basebackup/streamutil.c
***************
*** 83,89 **** GetConnection(void)
  	const char **keywords;
  	const char **values;
  	char	   *password = NULL;
- 	const char *tmpparam;
  
  	if (dbhost)
  		argcount++;
--- 83,88 ----
***************
*** 180,212 **** GetConnection(void)
  		free(values);
  		free(keywords);
  
- 		/*
- 		 * Ensure we have the same value of integer timestamps as the server
- 		 * we are connecting to.
- 		 */
- 		tmpparam = PQparameterStatus(tmpconn, "integer_datetimes");
- 		if (!tmpparam)
- 		{
- 			fprintf(stderr,
- 					_("%s: could not determine server setting for integer_datetimes\n"),
- 					progname);
- 			PQfinish(tmpconn);
- 			exit(1);
- 		}
- 
- #ifdef HAVE_INT64_TIMESTAMP
- 		if (strcmp(tmpparam, "on") != 0)
- #else
- 		if (strcmp(tmpparam, "off") != 0)
- #endif
- 		{
- 			fprintf(stderr,
- 			 _("%s: integer_datetimes compile flag does not match server\n"),
- 					progname);
- 			PQfinish(tmpconn);
- 			exit(1);
- 		}
- 
  		/* Store the password for next run */
  		if (password)
  			dbpassword = password;
--- 179,184 ----

*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 745,751 **** XLogWalRcvSendReply(bool force, bool requestReply)
--- 745,755 ----
  	pq_sendint64(&reply_message, writePtr);
  	pq_sendint64(&reply_message, flushPtr);
  	pq_sendint64(&reply_message, applyPtr);
+ #ifdef HAVE_INT64_TIMESTAMP
+ 	pq_sendint64(&reply_message, now);
+ #else
  	pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
+ #endif
  	pq_sendbyte(&reply_message, requestReply ? 1 : 0);
  
  	/* Send it */
***************
*** 816,822 **** XLogWalRcvSendHSFeedback(void)
--- 820,830 ----
  	/* Construct the the message and send it. */
  	resetStringInfo(&reply_message);
  	pq_sendbyte(&reply_message, 'h');
+ #ifdef HAVE_INT64_TIMESTAMP
+ 	pq_sendint64(&reply_message, now);
+ #else
  	pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
+ #endif
  	pq_sendint(&reply_message, xmin, 4);
  	pq_sendint(&reply_message, nextEpoch, 4);
  	walrcv_send(reply_message.data, reply_message.len);

*** a/src/bin/pg_basebackup/pg_basebackup.c
--- b/src/bin/pg_basebackup/pg_basebackup.c
***************
*** 47,52 **** bool		includewal = false;
--- 47,56 ----
  bool		streamwal = false;
  bool		fastcheckpoint = false;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int		standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char		*standby_connect_timeout = NULL;		
+ 
+ #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
  /* Progress counters */
  static uint64 totalsize;
***************
*** 125,130 **** usage(void)
--- 129,138 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));	
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
***************
*** 237,244 **** LogStreamerMain(logstreamer_param *param)
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout,
! 						   true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
--- 245,252 ----
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout, 
! 						   standby_recv_timeout, true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
***************
*** 290,297 **** StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier)
  	}
  #endif
  
! 	/* Get a second connection */
! 	param->bgconn = GetConnection();
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 298,306 ----
  	}
  #endif
  
! 	/* Get a second connection. Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.*/
! 	param->bgconn = GetConnection(standby_connect_timeout);
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
***************
*** 467,472 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 476,482 ----
  	char		filename[MAXPGPATH];
  	char	   *copybuf = NULL;
  	FILE	   *tarfile = NULL;
+ 	int64   last_recv_timestamp;
  
  #ifdef HAVE_LIBZ
  	gzFile		ztarfile = NULL;
***************
*** 584,592 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 594,605 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64		now;
  
  		if (copybuf != NULL)
  		{
***************
*** 594,600 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  		if (r == -1)
  		{
  			/*
--- 607,668 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
! 
! 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
! 			if (r == 0 || (r < 0 && errno == EINTR))
! 			{
! 				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 */
! 				if (standby_recv_timeout > 0)
! 				{
! 					now = localGetCurrentTimestamp();
! 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
! 					{
! 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
! 																			progname);
! 						disconnect_and_exit(1);
! 					}
! 				}
! 				
! 				continue;
! 			}
! 			else if (r < 0)
! 			{
! 				fprintf(stderr, _("%s: select() failed: %s\n"),
! 						progname, strerror(errno));
! 				disconnect_and_exit(1);
! 			}
! 			/* Else there is actually data on the socket */
! 			if (PQconsumeInput(conn) == 0)
! 			{
! 				fprintf(stderr,
! 						_("%s: could not receive data from WAL Sender: %s"),
! 						progname, PQerrorMessage(conn));
! 				disconnect_and_exit(1);
! 			}
! 
! 			/* Set the last reply timestamp */
! 			last_recv_timestamp = localGetCurrentTimestamp();
! 
! 			/* Some data is received, so go back read them in buffer*/			
! 			continue;
! 		}		
  		if (r == -1)
  		{
  			/*
***************
*** 665,670 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 733,741 ----
  			disconnect_and_exit(1);
  		}
  
+ 		/* Set the last reply timestamp */
+ 		last_recv_timestamp = localGetCurrentTimestamp();	
+ 
  #ifdef HAVE_LIBZ
  		if (ztarfile != NULL)
  		{
***************
*** 714,719 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
--- 785,791 ----
  	int			current_padding = 0;
  	char	   *copybuf = NULL;
  	FILE	   *file = NULL;
+ 	int64  last_recv_timestamp;
  
  	if (PQgetisnull(res, rownum, 0))
  		strcpy(current_path, basedir);
***************
*** 731,739 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
--- 803,815 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64    		now;
+ 
  
  		if (copybuf != NULL)
  		{
***************
*** 741,748 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  
  		if (r == -1)
  		{
  			/*
--- 817,878 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
  
+ 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
+ 			if (r == 0 || (r < 0 && errno == EINTR))
+ 			{
+ 				/*
+ 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
+ 				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
+ 																			progname);
+ 						disconnect_and_exit(1);
+ 					}
+ 				}
+ 				
+ 				continue;
+ 			}
+ 			else if (r < 0)
+ 			{
+ 				fprintf(stderr, _("%s: select() failed: %s\n"),
+ 						progname, strerror(errno));
+ 				disconnect_and_exit(1);
+ 			}
+ 			/* Else there is actually data on the socket */
+ 			if (PQconsumeInput(conn) == 0)
+ 			{
+ 				fprintf(stderr,
+ 						_("%s: could not receive data from WAL Sender: %s"),
+ 						progname, PQerrorMessage(conn));
+ 				disconnect_and_exit(1);
+ 			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 
+ 			/* Some data is received, so go back read them in buffer*/			
+ 			continue;
+ 		}
  		if (r == -1)
  		{
  			/*
***************
*** 755,765 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
  		}
  		else if (r == -2)
  		{
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 
  		if (file == NULL)
  		{
  			int			filemode;
--- 885,898 ----
  		}
  		else if (r == -2)
  		{
+ 		    fprintf(stderr, "\n");
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 		
! 		/* Set the last reply timestamp */
! 		last_recv_timestamp = localGetCurrentTimestamp();
  		if (file == NULL)
  		{
  			int			filemode;
***************
*** 953,961 **** BaseBackup(void)
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 1086,1094 ----
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server. Sending connect_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
***************
*** 1254,1259 **** main(int argc, char **argv)
--- 1387,1394 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},	
  		{"verbose", no_argument, NULL, 'v'},
  		{"progress", no_argument, NULL, 'P'},
  		{NULL, 0, NULL, 0}
***************
*** 1280,1286 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:xX:l:zZ:c:h:p:U:s:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 1415,1421 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:xX:l:zZ:c:h:p:U:s:r:t:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
***************
*** 1392,1397 **** main(int argc, char **argv)
--- 1527,1552 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 
+ 				break;			
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
  			case 'v':
  				verbose++;
  				break;
*** a/src/bin/pg_basebackup/pg_receivexlog.c
--- b/src/bin/pg_basebackup/pg_receivexlog.c
***************
*** 41,46 **** char	   *basedir = NULL;
--- 41,48 ----
  int			verbose = 0;
  int			noloop = 0;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int			standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char			*standby_connect_timeout = NULL;		
  volatile bool time_to_abort = false;
  
  
***************
*** 69,74 **** usage(void)
--- 71,80 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));		
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
***************
*** 224,232 **** StreamLog(void)
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
--- 230,239 ----
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server, Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
***************
*** 280,286 **** StreamLog(void)
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout, false);
  
  	PQfinish(conn);
  }
--- 287,294 ----
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout,
! 					  standby_recv_timeout, false);
  
  	PQfinish(conn);
  }
***************
*** 312,317 **** main(int argc, char **argv)
--- 320,327 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},				
  		{"verbose", no_argument, NULL, 'v'},
  		{NULL, 0, NULL, 0}
  	};
***************
*** 336,342 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 346,352 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:r:t:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
***************
*** 374,379 **** main(int argc, char **argv)
--- 384,409 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				break;
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
+ 				
  			case 'n':
  				noloop = 1;
  				break;
*** a/src/bin/pg_basebackup/receivelog.c
--- b/src/bin/pg_basebackup/receivelog.c
***************
*** 190,196 **** close_walfile(char *basedir, char *walname, bool segment_complete)
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! static int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
--- 190,196 ----
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
***************
*** 210,216 **** localGetCurrentTimestamp(void)
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! static void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
--- 210,216 ----
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
***************
*** 232,238 **** localTimestampDifference(int64 start_time, int64 stop_time,
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! static bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
--- 232,238 ----
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
***************
*** 342,348 **** bool
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
--- 342,349 ----
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, 
! 				  int standby_recv_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
***************
*** 350,355 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 351,358 ----
  	char	   *copybuf = NULL;
  	int64		last_status = -1;
  	XLogRecPtr	blockpos = InvalidXLogRecPtr;
+ 	int64	 last_recv_timestamp;
+ 	bool		ping_sent = false;
  
  	if (sysidentifier != NULL)
  	{
***************
*** 403,408 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 406,415 ----
  	}
  	PQclear(res);
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
+ 	ping_sent = false;
+ 	
  	/*
  	 * Receive the actual xlog data
  	 */
***************
*** 486,495 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Continue the loop and either
! 				 * deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
  				continue;
  			}
  			else if (r < 0)
--- 493,531 ----
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 * and then either deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating XLogStream receiver due to timeout\n"),
+ 																			progname);
+ 						goto error;
+ 					}
+ 
+ 					/*
+ 					 * We didn't receive anything new, for half of receiver
+ 					 * replication timeout. Ping the server, if not already done.
+ 					 */
+ 					if (!ping_sent)
+ 					{
+ 						if (localTimestampDifferenceExceeds(last_recv_timestamp, now, (standby_recv_timeout/2)))
+ 						{
+ 							if (!sendFeedback(conn, blockpos, now, true))
+ 							{
+ 								goto error;
+ 							}
+ 
+ 							last_status = now;
+ 							ping_sent = true;
+ 						}
+ 					}		
+ 				}			
+ 				
  				continue;
  			}
  			else if (r < 0)
***************
*** 506,511 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 542,552 ----
  						progname, PQerrorMessage(conn));
  				goto error;
  			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 			ping_sent = false;
+ 			
  			continue;
  		}
  		if (r == -1)
***************
*** 518,524 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  			goto error;
  		}
  
! 		/* Check the message type. */
  		if (copybuf[0] == 'k')
  		{
  			int		pos;
--- 559,568 ----
  			goto error;
  		}
  
! 		/* Set the last reply timestamp */
! 		last_recv_timestamp = localGetCurrentTimestamp();
! 		ping_sent = false;
! 		
  		if (copybuf[0] == 'k')
  		{
  			int		pos;
*** a/src/bin/pg_basebackup/receivelog.h
--- b/src/bin/pg_basebackup/receivelog.h
***************
*** 7,16 ****
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 				  XLogRecPtr startpos,
! 				  uint32 timeline,
! 				  char *sysidentifier,
! 				  char *basedir,
! 				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout,
! 				  bool rename_partial);
--- 7,24 ----
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 							  XLogRecPtr startpos,
! 							  uint32 timeline,
! 							  char *sysidentifier,
! 							  char *basedir,
! 							  stream_stop_callback stream_stop,
! 							  int standby_message_timeout,
! 							  int standby_recv_timeout,
! 							  bool rename_partial);
! 
! extern int64 localGetCurrentTimestamp(void);
! extern void  localTimestampDifference(int64 start_time, int64 stop_time,
! 								 long *secs, int *microsecs);
! extern bool  localTimestampDifferenceExceeds(int64 start_time,
! 								int64 stop_time,
! 								int msec);
*** a/src/bin/pg_basebackup/streamutil.c
--- b/src/bin/pg_basebackup/streamutil.c
***************
*** 72,80 **** pg_malloc0(size_t size)
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
   */
  PGconn *
! GetConnection(void)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
--- 72,82 ----
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
+  * Set conn_timeout to PGconn structure if their value 
+  * is not NULL.
   */
  PGconn *
! GetConnection(char *conn_timeout)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
***************
*** 91,96 **** GetConnection(void)
--- 93,100 ----
  		argcount++;
  	if (dbport)
  		argcount++;
+ 	if (conn_timeout)
+ 		argcount++;
  
  	keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
  	values = pg_malloc0((argcount + 1) * sizeof(*values));
***************
*** 120,125 **** GetConnection(void)
--- 124,135 ----
  		values[i] = dbport;
  		i++;
  	}
+ 	if (conn_timeout != NULL)
+ 	{
+ 		keywords[i] = "connect_timeout";
+ 		values[i] = conn_timeout;
+ 		i++;
+ 	}
  
  	while (true)
  	{
*** a/src/bin/pg_basebackup/streamutil.h
--- b/src/bin/pg_basebackup/streamutil.h
***************
*** 19,22 **** extern PGconn *conn;
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! extern PGconn *GetConnection(void);
--- 19,22 ----
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! PGconn	   *GetConnection(char *conn_timeout);

*** a/doc/src/sgml/ref/pg_basebackup.sgml
--- b/doc/src/sgml/ref/pg_basebackup.sgml
***************
*** 386,391 **** PostgreSQL documentation
--- 386,411 ----
       </varlistentry>
  
       <varlistentry>
+       <term><option>-r <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--recvtimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that receiver waits for communication from server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>     
+ 
+      <varlistentry>
+       <term><option>-t <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--conntimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that client wait for connection to establish with server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>          
+ 
+      <varlistentry>
        <term><option>-U <replaceable>username</replaceable></option></term>
        <term><option>--username=<replaceable class="parameter">username</replaceable></option></term>
        <listitem>
*** a/doc/src/sgml/ref/pg_receivexlog.sgml
--- b/doc/src/sgml/ref/pg_receivexlog.sgml
***************
*** 164,169 **** PostgreSQL documentation
--- 164,189 ----
       </varlistentry>
  
       <varlistentry>
+       <term><option>-r <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--recvtimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that receiver waits for communication from server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>     
+ 
+      <varlistentry>
+       <term><option>-t <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--conntimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that client wait for connection to establish with server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>  
+      
+      <varlistentry>
        <term><option>-U <replaceable>username</replaceable></option></term>
        <term><option>--username=<replaceable class="parameter">username</replaceable></option></term>
        <listitem>
*** a/src/bin/pg_basebackup/pg_basebackup.c
--- b/src/bin/pg_basebackup/pg_basebackup.c
***************
*** 41,46 **** bool		includewal = false;
--- 41,50 ----
  bool		streamwal = false;
  bool		fastcheckpoint = false;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int		standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char		*standby_connect_timeout = NULL;		
+ 
+ #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
  /* Progress counters */
  static uint64 totalsize;
***************
*** 119,124 **** usage(void)
--- 123,132 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));	
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
***************
*** 231,238 **** LogStreamerMain(logstreamer_param *param)
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout,
! 						   true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
--- 239,246 ----
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout, 
! 						   standby_recv_timeout, true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
***************
*** 284,291 **** StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier)
  	}
  #endif
  
! 	/* Get a second connection */
! 	param->bgconn = GetConnection();
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 292,300 ----
  	}
  #endif
  
! 	/* Get a second connection. Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.*/
! 	param->bgconn = GetConnection(standby_connect_timeout);
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
***************
*** 461,466 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 470,476 ----
  	char		filename[MAXPGPATH];
  	char	   *copybuf = NULL;
  	FILE	   *tarfile = NULL;
+ 	int64   last_recv_timestamp;
  
  #ifdef HAVE_LIBZ
  	gzFile		ztarfile = NULL;
***************
*** 578,586 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 588,599 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64		now;
  
  		if (copybuf != NULL)
  		{
***************
*** 588,594 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  		if (r == -1)
  		{
  			/*
--- 601,662 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
! 
! 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
! 			if (r == 0 || (r < 0 && errno == EINTR))
! 			{
! 				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 */
! 				if (standby_recv_timeout > 0)
! 				{
! 					now = localGetCurrentTimestamp();
! 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
! 					{
! 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
! 																			progname);
! 						disconnect_and_exit(1);
! 					}
! 				}
! 				
! 				continue;
! 			}
! 			else if (r < 0)
! 			{
! 				fprintf(stderr, _("%s: select() failed: %s\n"),
! 						progname, strerror(errno));
! 				disconnect_and_exit(1);
! 			}
! 			/* Else there is actually data on the socket */
! 			if (PQconsumeInput(conn) == 0)
! 			{
! 				fprintf(stderr,
! 						_("%s: could not receive data from WAL Sender: %s"),
! 						progname, PQerrorMessage(conn));
! 				disconnect_and_exit(1);
! 			}
! 
! 			/* Set the last reply timestamp */
! 			last_recv_timestamp = localGetCurrentTimestamp();
! 
! 			/* Some data is received, so go back read them in buffer*/			
! 			continue;
! 		}		
  		if (r == -1)
  		{
  			/*
***************
*** 659,664 **** ReceiveTarFile(PGconn *conn, PGresult *res, int rownum)
--- 727,735 ----
  			disconnect_and_exit(1);
  		}
  
+ 		/* Set the last reply timestamp */
+ 		last_recv_timestamp = localGetCurrentTimestamp();	
+ 
  #ifdef HAVE_LIBZ
  		if (ztarfile != NULL)
  		{
***************
*** 708,713 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
--- 779,785 ----
  	int			current_padding = 0;
  	char	   *copybuf = NULL;
  	FILE	   *file = NULL;
+ 	int64  last_recv_timestamp;
  
  	if (PQgetisnull(res, rownum, 0))
  		strcpy(current_path, basedir);
***************
*** 725,733 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
--- 797,809 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64    		now;
+ 
  
  		if (copybuf != NULL)
  		{
***************
*** 735,742 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  
  		if (r == -1)
  		{
  			/*
--- 811,872 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
  
+ 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
+ 			if (r == 0 || (r < 0 && errno == EINTR))
+ 			{
+ 				/*
+ 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
+ 				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
+ 																			progname);
+ 						disconnect_and_exit(1);
+ 					}
+ 				}
+ 				
+ 				continue;
+ 			}
+ 			else if (r < 0)
+ 			{
+ 				fprintf(stderr, _("%s: select() failed: %s\n"),
+ 						progname, strerror(errno));
+ 				disconnect_and_exit(1);
+ 			}
+ 			/* Else there is actually data on the socket */
+ 			if (PQconsumeInput(conn) == 0)
+ 			{
+ 				fprintf(stderr,
+ 						_("%s: could not receive data from WAL Sender: %s"),
+ 						progname, PQerrorMessage(conn));
+ 				disconnect_and_exit(1);
+ 			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 
+ 			/* Some data is received, so go back read them in buffer*/			
+ 			continue;
+ 		}
  		if (r == -1)
  		{
  			/*
***************
*** 749,759 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
  		}
  		else if (r == -2)
  		{
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 
  		if (file == NULL)
  		{
  			int			filemode;
--- 879,892 ----
  		}
  		else if (r == -2)
  		{
+ 		    fprintf(stderr, "\n");
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 		
! 		/* Set the last reply timestamp */
! 		last_recv_timestamp = localGetCurrentTimestamp();
  		if (file == NULL)
  		{
  			int			filemode;
***************
*** 947,955 **** BaseBackup(void)
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 1080,1088 ----
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server. Sending connect_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
***************
*** 1248,1253 **** main(int argc, char **argv)
--- 1381,1388 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},	
  		{"verbose", no_argument, NULL, 'v'},
  		{"progress", no_argument, NULL, 'P'},
  		{NULL, 0, NULL, 0}
***************
*** 1274,1280 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:xX:l:zZ:c:h:p:U:s:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 1409,1415 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:xX:l:zZ:c:h:p:U:s:r:t:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
***************
*** 1386,1391 **** main(int argc, char **argv)
--- 1521,1546 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 
+ 				break;			
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
  			case 'v':
  				verbose++;
  				break;
*** a/src/bin/pg_basebackup/pg_receivexlog.c
--- b/src/bin/pg_basebackup/pg_receivexlog.c
***************
*** 35,40 **** char	   *basedir = NULL;
--- 35,42 ----
  int			verbose = 0;
  int			noloop = 0;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int			standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char			*standby_connect_timeout = NULL;		
  volatile bool time_to_abort = false;
  
  
***************
*** 63,68 **** usage(void)
--- 65,74 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));		
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
***************
*** 218,226 **** StreamLog(void)
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
--- 224,233 ----
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server, Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
***************
*** 274,280 **** StreamLog(void)
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout, false);
  
  	PQfinish(conn);
  }
--- 281,288 ----
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout,
! 					  standby_recv_timeout, false);
  
  	PQfinish(conn);
  }
***************
*** 306,311 **** main(int argc, char **argv)
--- 314,321 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},				
  		{"verbose", no_argument, NULL, 'v'},
  		{NULL, 0, NULL, 0}
  	};
***************
*** 331,337 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 341,347 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:r:t:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
***************
*** 369,374 **** main(int argc, char **argv)
--- 379,404 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				break;
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
+ 				
  			case 'n':
  				noloop = 1;
  				break;
*** a/src/bin/pg_basebackup/receivelog.c
--- b/src/bin/pg_basebackup/receivelog.c
***************
*** 181,187 **** close_walfile(char *basedir, char *walname, bool segment_complete)
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! static int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
--- 181,187 ----
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
***************
*** 201,207 **** localGetCurrentTimestamp(void)
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! static void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
--- 201,207 ----
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
***************
*** 223,229 **** localTimestampDifference(int64 start_time, int64 stop_time,
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! static bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
--- 223,229 ----
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
***************
*** 333,339 **** bool
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
--- 333,340 ----
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, 
! 				  int standby_recv_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
***************
*** 341,346 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 342,349 ----
  	char	   *copybuf = NULL;
  	int64		last_status = -1;
  	XLogRecPtr	blockpos = InvalidXLogRecPtr;
+ 	int64	 last_recv_timestamp;
+ 	bool		ping_sent = false;
  
  	if (sysidentifier != NULL)
  	{
***************
*** 394,399 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 397,406 ----
  	}
  	PQclear(res);
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
+ 	ping_sent = false;
+ 	
  	/*
  	 * Receive the actual xlog data
  	 */
***************
*** 477,486 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Continue the loop and either
! 				 * deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
  				continue;
  			}
  			else if (r < 0)
--- 484,522 ----
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 * and then either deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating XLogStream receiver due to timeout\n"),
+ 																			progname);
+ 						goto error;
+ 					}
+ 
+ 					/*
+ 					 * We didn't receive anything new, for half of receiver
+ 					 * replication timeout. Ping the server, if not already done.
+ 					 */
+ 					if (!ping_sent)
+ 					{
+ 						if (localTimestampDifferenceExceeds(last_recv_timestamp, now, (standby_recv_timeout/2)))
+ 						{
+ 							if (!sendFeedback(conn, blockpos, now, true))
+ 							{
+ 								goto error;
+ 							}
+ 
+ 							last_status = now;
+ 							ping_sent = true;
+ 						}
+ 					}		
+ 				}			
+ 				
  				continue;
  			}
  			else if (r < 0)
***************
*** 497,502 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 533,543 ----
  						progname, PQerrorMessage(conn));
  				goto error;
  			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 			ping_sent = false;
+ 			
  			continue;
  		}
  		if (r == -1)
***************
*** 509,514 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
--- 550,559 ----
  			goto error;
  		}
  
+ 		/* Set the last reply timestamp */
+ 		last_recv_timestamp = localGetCurrentTimestamp();
+ 		ping_sent = false;
+ 
  		/* Check the message type. */
  		if (copybuf[0] == 'k')
  		{
*** a/src/bin/pg_basebackup/receivelog.h
--- b/src/bin/pg_basebackup/receivelog.h
***************
*** 7,16 ****
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 				  XLogRecPtr startpos,
! 				  uint32 timeline,
! 				  char *sysidentifier,
! 				  char *basedir,
! 				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout,
! 				  bool rename_partial);
--- 7,24 ----
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 							  XLogRecPtr startpos,
! 							  uint32 timeline,
! 							  char *sysidentifier,
! 							  char *basedir,
! 							  stream_stop_callback stream_stop,
! 							  int standby_message_timeout,
! 							  int standby_recv_timeout,
! 							  bool rename_partial);
! 
! extern int64 localGetCurrentTimestamp(void);
! extern void  localTimestampDifference(int64 start_time, int64 stop_time,
! 								 long *secs, int *microsecs);
! extern bool  localTimestampDifferenceExceeds(int64 start_time,
! 								int64 stop_time,
! 								int msec);
*** a/src/bin/pg_basebackup/streamutil.c
--- b/src/bin/pg_basebackup/streamutil.c
***************
*** 66,74 **** pg_malloc0(size_t size)
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
   */
  PGconn *
! GetConnection(void)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
--- 66,76 ----
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
+  * Set conn_timeout to PGconn structure if their value 
+  * is not NULL.
   */
  PGconn *
! GetConnection(char *conn_timeout)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
***************
*** 85,90 **** GetConnection(void)
--- 87,94 ----
  		argcount++;
  	if (dbport)
  		argcount++;
+ 	if (conn_timeout)
+ 		argcount++;
  
  	keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
  	values = pg_malloc0((argcount + 1) * sizeof(*values));
***************
*** 114,119 **** GetConnection(void)
--- 118,129 ----
  		values[i] = dbport;
  		i++;
  	}
+ 	if (conn_timeout != NULL)
+ 	{
+ 		keywords[i] = "connect_timeout";
+ 		values[i] = conn_timeout;
+ 		i++;
+ 	}
  
  	while (true)
  	{
*** a/src/bin/pg_basebackup/streamutil.h
--- b/src/bin/pg_basebackup/streamutil.h
***************
*** 19,22 **** extern PGconn *conn;
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! extern PGconn *GetConnection(void);
--- 19,22 ----
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! PGconn	   *GetConnection(char *conn_timeout);

diff -dcrpN postgresql.orig/doc/src/sgml/ref/pg_basebackup.sgml postgresql/doc/src/sgml/ref/pg_basebackup.sgml
*** postgresql.orig/doc/src/sgml/ref/pg_basebackup.sgml	2013-01-05 17:34:30.742135371 +0100
--- postgresql/doc/src/sgml/ref/pg_basebackup.sgml	2013-01-07 15:11:40.787007890 +0100
*************** PostgreSQL documentation
*** 400,405 ****
--- 400,425 ----
       </varlistentry>
  
       <varlistentry>
+       <term><option>-r <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--recvtimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that receiver waits for communication from server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>     
+ 
+      <varlistentry>
+       <term><option>-t <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--conntimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that client wait for connection to establish with server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>          
+ 
+      <varlistentry>
        <term><option>-U <replaceable>username</replaceable></option></term>
        <term><option>--username=<replaceable class="parameter">username</replaceable></option></term>
        <listitem>
diff -dcrpN postgresql.orig/doc/src/sgml/ref/pg_receivexlog.sgml postgresql/doc/src/sgml/ref/pg_receivexlog.sgml
*** postgresql.orig/doc/src/sgml/ref/pg_receivexlog.sgml	2012-11-08 13:13:04.152630639 +0100
--- postgresql/doc/src/sgml/ref/pg_receivexlog.sgml	2013-01-07 15:11:40.788007898 +0100
*************** PostgreSQL documentation
*** 164,169 ****
--- 164,189 ----
       </varlistentry>
  
       <varlistentry>
+       <term><option>-r <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--recvtimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that receiver waits for communication from server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>     
+ 
+      <varlistentry>
+       <term><option>-t <replaceable class="parameter">interval</replaceable></option></term>
+       <term><option>--conntimeout=<replaceable class="parameter">interval</replaceable></option></term>
+       <listitem>
+        <para>
+         time that client wait for connection to establish with server (in seconds).
+        </para>
+       </listitem>
+      </varlistentry>  
+      
+      <varlistentry>
        <term><option>-U <replaceable>username</replaceable></option></term>
        <term><option>--username=<replaceable class="parameter">username</replaceable></option></term>
        <listitem>
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/pg_basebackup.c postgresql/src/bin/pg_basebackup/pg_basebackup.c
*** postgresql.orig/src/bin/pg_basebackup/pg_basebackup.c	2013-01-05 17:34:30.778135625 +0100
--- postgresql/src/bin/pg_basebackup/pg_basebackup.c	2013-01-07 15:16:24.610037886 +0100
*************** bool		streamwal = false;
*** 45,50 ****
--- 45,54 ----
  bool		fastcheckpoint = false;
  bool		writerecoveryconf = false;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int		standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char		*standby_connect_timeout = NULL;		
+ 
+ #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
  /* Progress counters */
  static uint64 totalsize;
*************** usage(void)
*** 130,135 ****
--- 134,143 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));	
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
*************** LogStreamerMain(logstreamer_param *param
*** 242,249 ****
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout,
! 						   true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
--- 250,257 ----
  {
  	if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline,
  						   param->sysidentifier, param->xlogdir,
! 						   reached_end_position, standby_message_timeout, 
! 						   standby_recv_timeout, true))
  
  		/*
  		 * Any errors will already have been reported in the function process,
*************** StartLogStreamer(char *startpos, uint32
*** 295,302 ****
  	}
  #endif
  
! 	/* Get a second connection */
! 	param->bgconn = GetConnection();
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 303,311 ----
  	}
  #endif
  
! 	/* Get a second connection. Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.*/
! 	param->bgconn = GetConnection(standby_connect_timeout);
  	if (!param->bgconn)
  		/* Error message already written in GetConnection() */
  		exit(1);
*************** ReceiveTarFile(PGconn *conn, PGresult *r
*** 511,516 ****
--- 520,526 ----
  	char		filename[MAXPGPATH];
  	char	   *copybuf = NULL;
  	FILE	   *tarfile = NULL;
+ 	int64		last_recv_timestamp;
  	char		tarhdr[512];
  	bool		basetablespace = PQgetisnull(res, rownum, 0);
  	bool		in_tarhdr = true;
*************** ReceiveTarFile(PGconn *conn, PGresult *r
*** 634,642 ****
--- 644,655 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64		now;
  
  		if (copybuf != NULL)
  		{
*************** ReceiveTarFile(PGconn *conn, PGresult *r
*** 644,650 ****
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  		if (r == -1)
  		{
  			/*
--- 657,718 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
! 
! 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
! 			if (r == 0 || (r < 0 && errno == EINTR))
! 			{
! 				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 */
! 				if (standby_recv_timeout > 0)
! 				{
! 					now = localGetCurrentTimestamp();
! 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
! 					{
! 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
! 																			progname);
! 						disconnect_and_exit(1);
! 					}
! 				}
! 				
! 				continue;
! 			}
! 			else if (r < 0)
! 			{
! 				fprintf(stderr, _("%s: select() failed: %s\n"),
! 						progname, strerror(errno));
! 				disconnect_and_exit(1);
! 			}
! 			/* Else there is actually data on the socket */
! 			if (PQconsumeInput(conn) == 0)
! 			{
! 				fprintf(stderr,
! 						_("%s: could not receive data from WAL Sender: %s"),
! 						progname, PQerrorMessage(conn));
! 				disconnect_and_exit(1);
! 			}
! 
! 			/* Set the last reply timestamp */
! 			last_recv_timestamp = localGetCurrentTimestamp();
! 
! 			/* Some data is received, so go back read them in buffer*/			
! 			continue;
! 		}		
  		if (r == -1)
  		{
  			/*
*************** ReceiveTarFile(PGconn *conn, PGresult *r
*** 680,685 ****
--- 748,756 ----
  			/* 2 * 512 bytes empty data at end of file */
  			WRITE_TAR_DATA(zerobuf, sizeof(zerobuf));
  
+ 		/* Set the last reply timestamp */
+ 		last_recv_timestamp = localGetCurrentTimestamp();	
+ 
  #ifdef HAVE_LIBZ
  			if (ztarfile != NULL)
  			{
*************** ReceiveAndUnpackTarFile(PGconn *conn, PG
*** 860,865 ****
--- 931,937 ----
  	bool		basetablespace = PQgetisnull(res, rownum, 0);
  	char	   *copybuf = NULL;
  	FILE	   *file = NULL;
+ 	int64  last_recv_timestamp;
  
  	if (basetablespace)
  		strcpy(current_path, basedir);
*************** ReceiveAndUnpackTarFile(PGconn *conn, PG
*** 877,885 ****
--- 949,961 ----
  		disconnect_and_exit(1);
  	}
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
  	while (1)
  	{
  		int			r;
+ 		int64    		now;
+ 
  
  		if (copybuf != NULL)
  		{
*************** ReceiveAndUnpackTarFile(PGconn *conn, PG
*** 887,894 ****
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 0);
  
  		if (r == -1)
  		{
  			/*
--- 963,1024 ----
  			copybuf = NULL;
  		}
  
! 		r = PQgetCopyData(conn, &copybuf, 1);
! 		if (r == 0)
! 		{
! 			/*
! 			 * In async mode, and no data available. We block on reading but
! 			 * not more than the specified timeout, so that we can send a
! 			 * response back to the client.
! 			 */
! 			fd_set		input_mask;
! 			struct timeval timeout;
! 
! 			FD_ZERO(&input_mask);
! 			FD_SET(PQsocket(conn), &input_mask);
! 			timeout.tv_sec = 0; 
! 			timeout.tv_usec = NAPTIME_PER_CYCLE*1000;
  
+ 			r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, &timeout);
+ 			if (r == 0 || (r < 0 && errno == EINTR))
+ 			{
+ 				/*
+ 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
+ 				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating DB File receive due to timeout\n"),
+ 																			progname);
+ 						disconnect_and_exit(1);
+ 					}
+ 				}
+ 				
+ 				continue;
+ 			}
+ 			else if (r < 0)
+ 			{
+ 				fprintf(stderr, _("%s: select() failed: %s\n"),
+ 						progname, strerror(errno));
+ 				disconnect_and_exit(1);
+ 			}
+ 			/* Else there is actually data on the socket */
+ 			if (PQconsumeInput(conn) == 0)
+ 			{
+ 				fprintf(stderr,
+ 						_("%s: could not receive data from WAL Sender: %s"),
+ 						progname, PQerrorMessage(conn));
+ 				disconnect_and_exit(1);
+ 			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 
+ 			/* Some data is received, so go back read them in buffer*/			
+ 			continue;
+ 		}
  		if (r == -1)
  		{
  			/*
*************** ReceiveAndUnpackTarFile(PGconn *conn, PG
*** 901,911 ****
  		}
  		else if (r == -2)
  		{
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 
  		if (file == NULL)
  		{
  			int			filemode;
--- 1031,1044 ----
  		}
  		else if (r == -2)
  		{
+ 		    fprintf(stderr, "\n");
  			fprintf(stderr, _("%s: could not read COPY data: %s"),
  					progname, PQerrorMessage(conn));
  			disconnect_and_exit(1);
  		}
! 		
! 		/* Set the last reply timestamp */
! 		last_recv_timestamp = localGetCurrentTimestamp();
  		if (file == NULL)
  		{
  			int			filemode;
*************** BaseBackup(void)
*** 1213,1221 ****
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
--- 1346,1354 ----
  	char		xlogend[64];
  
  	/*
! 	 * Connect in replication mode to the server. Sending connect_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		exit(1);
*************** main(int argc, char **argv)
*** 1524,1529 ****
--- 1657,1664 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},	
  		{"verbose", no_argument, NULL, 'v'},
  		{"progress", no_argument, NULL, 'P'},
  		{NULL, 0, NULL, 0}
*************** main(int argc, char **argv)
*** 1550,1556 ****
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:RxX:l:zZ:c:h:p:U:s:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 1685,1691 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:RxX:l:zZ:c:h:p:U:s:r:t:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
*************** main(int argc, char **argv)
*** 1665,1670 ****
--- 1800,1825 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 
+ 				break;			
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
  			case 'v':
  				verbose++;
  				break;
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/pg_receivexlog.c postgresql/src/bin/pg_basebackup/pg_receivexlog.c
*** postgresql.orig/src/bin/pg_basebackup/pg_receivexlog.c	2013-01-02 09:19:03.856521815 +0100
--- postgresql/src/bin/pg_basebackup/pg_receivexlog.c	2013-01-07 15:11:40.792007931 +0100
*************** char	   *basedir = NULL;
*** 35,40 ****
--- 35,42 ----
  int			verbose = 0;
  int			noloop = 0;
  int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
+ int			standby_recv_timeout = 60*1000;		/* 60 sec = default */
+ char			*standby_connect_timeout = NULL;		
  volatile bool time_to_abort = false;
  
  
*************** usage(void)
*** 63,68 ****
--- 65,74 ----
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
  			 "                         time between status packets sent to server (in seconds)\n"));
+ 	printf(_("  -r, --recvtimeout=INTERVAL time that receiver waits for communication from\n"
+ 		   "                             server (in seconds)\n"));
+ 	printf(_("  -t, --conntimeout=INTERVAL time that client wait for connection to establish\n"
+ 		   "                             with server (in seconds)\n"));		
  	printf(_("  -U, --username=NAME    connect as specified database user\n"));
  	printf(_("  -w, --no-password      never prompt for password\n"));
  	printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
*************** StreamLog(void)
*** 218,226 ****
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server
  	 */
! 	conn = GetConnection();
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
--- 224,233 ----
  				lo;
  
  	/*
! 	 * Connect in replication mode to the server, Sending connect_timeout
! 	 * as configured, there is no need for rw_timeout.
  	 */
! 	conn = GetConnection(standby_connect_timeout);
  	if (!conn)
  		/* Error message already written in GetConnection() */
  		return;
*************** StreamLog(void)
*** 274,280 ****
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout, false);
  
  	PQfinish(conn);
  }
--- 281,288 ----
  				timeline);
  
  	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
! 					  stop_streaming, standby_message_timeout,
! 					  standby_recv_timeout, false);
  
  	PQfinish(conn);
  }
*************** main(int argc, char **argv)
*** 306,311 ****
--- 314,321 ----
  		{"no-password", no_argument, NULL, 'w'},
  		{"password", no_argument, NULL, 'W'},
  		{"status-interval", required_argument, NULL, 's'},
+ 		{"recvtimeout", required_argument, NULL, 'r'},		
+ 		{"conntimeout", required_argument, NULL, 't'},				
  		{"verbose", no_argument, NULL, 'v'},
  		{NULL, 0, NULL, 0}
  	};
*************** main(int argc, char **argv)
*** 331,337 ****
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 341,347 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:r:t:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
*************** main(int argc, char **argv)
*** 369,374 ****
--- 379,404 ----
  					exit(1);
  				}
  				break;
+ 			case 'r':
+ 				standby_recv_timeout = atoi(optarg)*1000;
+ 				if (standby_recv_timeout < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid recv timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				break;
+ 			case 't':
+ 				if (atoi(optarg) < 0)
+ 				{
+ 					fprintf(stderr, _("%s: invalid connect timeout \"%s\"\n"),
+ 							progname, optarg);
+ 					exit(1);
+ 				}
+ 				
+ 				standby_connect_timeout = pg_strdup(optarg);
+ 				break;								
+ 				
  			case 'n':
  				noloop = 1;
  				break;
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/receivelog.c postgresql/src/bin/pg_basebackup/receivelog.c
*** postgresql.orig/src/bin/pg_basebackup/receivelog.c	2013-01-02 09:19:03.856521815 +0100
--- postgresql/src/bin/pg_basebackup/receivelog.c	2013-01-07 15:11:40.792007931 +0100
*************** close_walfile(char *basedir, char *walna
*** 181,187 ****
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! static int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
--- 181,187 ----
   * backend code. The protocol always uses integer timestamps, regardless of
   * server setting.
   */
! int64
  localGetCurrentTimestamp(void)
  {
  	int64 result;
*************** localGetCurrentTimestamp(void)
*** 201,207 ****
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! static void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
--- 201,207 ----
   * Local version of TimestampDifference(), since we are not linked with
   * backend code.
   */
! void
  localTimestampDifference(int64 start_time, int64 stop_time,
  						 long *secs, int *microsecs)
  {
*************** localTimestampDifference(int64 start_tim
*** 223,229 ****
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! static bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
--- 223,229 ----
   * Local version of TimestampDifferenceExceeds(), since we are not
   * linked with backend code.
   */
! bool
  localTimestampDifferenceExceeds(int64 start_time,
  								int64 stop_time,
  								int msec)
*************** bool
*** 333,339 ****
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
--- 333,340 ----
  ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
  				  char *sysidentifier, char *basedir,
  				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout, 
! 				  int standby_recv_timeout, bool rename_partial)
  {
  	char		query[128];
  	char		current_walfile_name[MAXPGPATH];
*************** ReceiveXlogStream(PGconn *conn, XLogRecP
*** 341,346 ****
--- 342,349 ----
  	char	   *copybuf = NULL;
  	int64		last_status = -1;
  	XLogRecPtr	blockpos = InvalidXLogRecPtr;
+ 	int64	 last_recv_timestamp;
+ 	bool		ping_sent = false;
  
  	if (sysidentifier != NULL)
  	{
*************** ReceiveXlogStream(PGconn *conn, XLogRecP
*** 394,399 ****
--- 397,406 ----
  	}
  	PQclear(res);
  
+ 	/* Set the last reply timestamp */
+ 	last_recv_timestamp = localGetCurrentTimestamp();
+ 	ping_sent = false;
+ 	
  	/*
  	 * Receive the actual xlog data
  	 */
*************** ReceiveXlogStream(PGconn *conn, XLogRecP
*** 477,486 ****
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Continue the loop and either
! 				 * deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
  				continue;
  			}
  			else if (r < 0)
--- 484,522 ----
  			if (r == 0 || (r < 0 && errno == EINTR))
  			{
  				/*
! 				 * Got a timeout or signal. Before Continuing the loop, check for timeout.
! 				 * and then either deliver a status packet to the server or just go back into
  				 * blocking.
  				 */
+ 				if (standby_recv_timeout > 0)
+ 				{
+ 					now = localGetCurrentTimestamp();
+ 					if (localTimestampDifferenceExceeds(last_recv_timestamp, now, standby_recv_timeout))
+ 					{
+ 						fprintf(stderr, _("%s: terminating XLogStream receiver due to timeout\n"),
+ 																			progname);
+ 						goto error;
+ 					}
+ 
+ 					/*
+ 					 * We didn't receive anything new, for half of receiver
+ 					 * replication timeout. Ping the server, if not already done.
+ 					 */
+ 					if (!ping_sent)
+ 					{
+ 						if (localTimestampDifferenceExceeds(last_recv_timestamp, now, (standby_recv_timeout/2)))
+ 						{
+ 							if (!sendFeedback(conn, blockpos, now, true))
+ 							{
+ 								goto error;
+ 							}
+ 
+ 							last_status = now;
+ 							ping_sent = true;
+ 						}
+ 					}		
+ 				}			
+ 				
  				continue;
  			}
  			else if (r < 0)
*************** ReceiveXlogStream(PGconn *conn, XLogRecP
*** 497,502 ****
--- 533,543 ----
  						progname, PQerrorMessage(conn));
  				goto error;
  			}
+ 
+ 			/* Set the last reply timestamp */
+ 			last_recv_timestamp = localGetCurrentTimestamp();
+ 			ping_sent = false;
+ 			
  			continue;
  		}
  		if (r == -1)
*************** ReceiveXlogStream(PGconn *conn, XLogRecP
*** 509,514 ****
--- 550,559 ----
  			goto error;
  		}
  
+ 		/* Set the last reply timestamp */
+ 		last_recv_timestamp = localGetCurrentTimestamp();
+ 		ping_sent = false;
+ 
  		/* Check the message type. */
  		if (copybuf[0] == 'k')
  		{
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/receivelog.h postgresql/src/bin/pg_basebackup/receivelog.h
*** postgresql.orig/src/bin/pg_basebackup/receivelog.h	2012-06-11 06:22:48.200921787 +0200
--- postgresql/src/bin/pg_basebackup/receivelog.h	2013-01-07 15:11:40.793007938 +0100
***************
*** 7,16 ****
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 				  XLogRecPtr startpos,
! 				  uint32 timeline,
! 				  char *sysidentifier,
! 				  char *basedir,
! 				  stream_stop_callback stream_stop,
! 				  int standby_message_timeout,
! 				  bool rename_partial);
--- 7,24 ----
  typedef bool (*stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished);
  
  extern bool ReceiveXlogStream(PGconn *conn,
! 							  XLogRecPtr startpos,
! 							  uint32 timeline,
! 							  char *sysidentifier,
! 							  char *basedir,
! 							  stream_stop_callback stream_stop,
! 							  int standby_message_timeout,
! 							  int standby_recv_timeout,
! 							  bool rename_partial);
! 
! extern int64 localGetCurrentTimestamp(void);
! extern void  localTimestampDifference(int64 start_time, int64 stop_time,
! 								 long *secs, int *microsecs);
! extern bool  localTimestampDifferenceExceeds(int64 start_time,
! 								int64 stop_time,
! 								int msec);
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/streamutil.c postgresql/src/bin/pg_basebackup/streamutil.c
*** postgresql.orig/src/bin/pg_basebackup/streamutil.c	2013-01-02 09:19:03.856521815 +0100
--- postgresql/src/bin/pg_basebackup/streamutil.c	2013-01-07 15:11:40.793007938 +0100
*************** pg_malloc0(size_t size)
*** 66,74 ****
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
   */
  PGconn *
! GetConnection(void)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
--- 66,76 ----
   * Connect to the server. Returns a valid PGconn pointer if connected,
   * or NULL on non-permanent error. On permanent error, the function will
   * call exit(1) directly.
+  * Set conn_timeout to PGconn structure if their value 
+  * is not NULL.
   */
  PGconn *
! GetConnection(char *conn_timeout)
  {
  	PGconn	   *tmpconn;
  	int			argcount = 4;	/* dbname, replication, fallback_app_name,
*************** GetConnection(void)
*** 85,90 ****
--- 87,94 ----
  		argcount++;
  	if (dbport)
  		argcount++;
+ 	if (conn_timeout)
+ 		argcount++;
  
  	keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
  	values = pg_malloc0((argcount + 1) * sizeof(*values));
*************** GetConnection(void)
*** 114,119 ****
--- 118,129 ----
  		values[i] = dbport;
  		i++;
  	}
+ 	if (conn_timeout != NULL)
+ 	{
+ 		keywords[i] = "connect_timeout";
+ 		values[i] = conn_timeout;
+ 		i++;
+ 	}
  
  	while (true)
  	{
diff -dcrpN postgresql.orig/src/bin/pg_basebackup/streamutil.h postgresql/src/bin/pg_basebackup/streamutil.h
*** postgresql.orig/src/bin/pg_basebackup/streamutil.h	2012-10-03 10:40:48.299207401 +0200
--- postgresql/src/bin/pg_basebackup/streamutil.h	2013-01-07 15:11:40.794007945 +0100
*************** extern PGconn *conn;
*** 19,22 ****
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! extern PGconn *GetConnection(void);
--- 19,22 ----
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
  
! PGconn	   *GetConnection(char *conn_timeout);

*** a/doc/src/sgml/ref/pg_basebackup.sgml
--- b/doc/src/sgml/ref/pg_basebackup.sgml
***************
*** 357,362 **** PostgreSQL documentation
--- 357,375 ----
     <para>
      The following command-line options control the database connection parameters.
  
+ 	<variablelist>
+      <varlistentry>
+       <term><option>-C <replaceable class="parameter">"OPTIONS"</replaceable></option></term>
+       <term><option>--connection-string=<replaceable class="parameter">"OPTIONS"</replaceable></option></term>
+       <listitem>
+        <para>
+         Specifies connection string options, used for connecting to server.
+         These option can be used along with other user supplied options.
+         In case of conflicting option user supplied option is choosen.
+        </para>
+       </listitem>
+      </varlistentry>
+ 
      <variablelist>
       <varlistentry>
        <term><option>-h <replaceable class="parameter">host</replaceable></option></term>
*** a/doc/src/sgml/ref/pg_receivexlog.sgml
--- b/doc/src/sgml/ref/pg_receivexlog.sgml
***************
*** 121,126 **** PostgreSQL documentation
--- 121,139 ----
     <para>
      The following command-line options control the database connection parameters.
  
+ 	<variablelist>
+      <varlistentry>
+       <term><option>-C <replaceable class="parameter">"OPTIONS"</replaceable></option></term>
+       <term><option>--connection-string=<replaceable class="parameter">"OPTIONS"</replaceable></option></term>
+       <listitem>
+        <para>
+         Specifies connection string options, used for connecting to server.
+         These option can be used along with other user supplied options.
+         In case of conflicting option user supplied option is choosen.
+        </para>
+       </listitem>
+      </varlistentry>
+ 
      <variablelist>
       <varlistentry>
        <term><option>-h <replaceable class="parameter">host</replaceable></option></term>
*** a/src/bin/pg_basebackup/pg_basebackup.c
--- b/src/bin/pg_basebackup/pg_basebackup.c
***************
*** 126,131 **** usage(void)
--- 126,132 ----
  	printf(_("  -V, --version          output version information, then exit\n"));
  	printf(_("  -?, --help             show this help, then exit\n"));
  	printf(_("\nConnection options:\n"));
+ 	printf(_("  -C, --connection-string=\"OPTIONS\" Connect to server using OPTIONS\n"));
  	printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
***************
*** 1104,1125 **** ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum)
  }
  
  /*
-  * Escape single quotes used in connection parameters
-  */
- static char *
- escape_quotes(const char *src)
- {
- 	char	   *result = escape_single_quotes_ascii(src);
- 
- 	if (!result)
- 	{
- 		fprintf(stderr, _("%s: out of memory\n"), progname);
- 		exit(1);
- 	}
- 	return result;
- }
- 
- /*
   * Create a recovery.conf file in memory using a PQExpBuffer
   */
  static void
--- 1105,1110 ----
***************
*** 1532,1537 **** main(int argc, char **argv)
--- 1517,1523 ----
  		{"pgdata", required_argument, NULL, 'D'},
  		{"format", required_argument, NULL, 'F'},
  		{"checkpoint", required_argument, NULL, 'c'},
+ 		{"connection-string", required_argument, NULL, 'C'},
  		{"write-recovery-conf", no_argument, NULL, 'R'},
  		{"xlog", no_argument, NULL, 'x'},
  		{"xlog-method", required_argument, NULL, 'X'},
***************
*** 1570,1576 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:RxX:l:zZ:c:h:p:U:s:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
--- 1556,1562 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:F:RxX:l:zZ:C:c:h:p:U:s:wWvP",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
***************
*** 1661,1666 **** main(int argc, char **argv)
--- 1647,1655 ----
  					exit(1);
  				}
  				break;
+ 			case 'C':
+ 				connection_string = pg_strdup(optarg);
+ 				break;
  			case 'h':
  				dbhost = pg_strdup(optarg);
  				break;
*** a/src/bin/pg_basebackup/pg_receivexlog.c
--- b/src/bin/pg_basebackup/pg_receivexlog.c
***************
*** 58,63 **** usage(void)
--- 58,64 ----
  	printf(_("  -V, --version          output version information, then exit\n"));
  	printf(_("  -?, --help             show this help, then exit\n"));
  	printf(_("\nConnection options:\n"));
+ 	printf(_("  -C, --connection-string=\"OPTIONS\" Connect to server using OPTIONS\n"));
  	printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
  	printf(_("  -p, --port=PORT        database server port number\n"));
  	printf(_("  -s, --status-interval=INTERVAL\n"
***************
*** 305,310 **** main(int argc, char **argv)
--- 306,312 ----
  	static struct option long_options[] = {
  		{"help", no_argument, NULL, '?'},
  		{"version", no_argument, NULL, 'V'},
+ 		{"connection-string", required_argument, NULL, 'C'},
  		{"directory", required_argument, NULL, 'D'},
  		{"host", required_argument, NULL, 'h'},
  		{"port", required_argument, NULL, 'p'},
***************
*** 338,348 **** main(int argc, char **argv)
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "D:h:p:U:s:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
  		{
  			case 'D':
  				basedir = pg_strdup(optarg);
  				break;
--- 340,353 ----
  		}
  	}
  
! 	while ((c = getopt_long(argc, argv, "C:D:h:p:U:s:nwWv",
  							long_options, &option_index)) != -1)
  	{
  		switch (c)
  		{
+ 			case 'C':
+ 				connection_string = pg_strdup(optarg);
+ 				break;
  			case 'D':
  				basedir = pg_strdup(optarg);
  				break;
*** a/src/bin/pg_basebackup/streamutil.c
--- b/src/bin/pg_basebackup/streamutil.c
***************
*** 24,29 **** char	   *dbport = NULL;
--- 24,47 ----
  int			dbgetpassword = 0;	/* 0=auto, -1=never, 1=always */
  static char *dbpassword = NULL;
  PGconn	   *conn = NULL;
+ char	   *connection_string = NULL;
+ 
+ 
+ /*
+  * Escape single quotes used in connection parameters
+  */
+ char *
+ escape_quotes(const char *src)
+ {
+ 	char	   *result = escape_single_quotes_ascii(src);
+ 
+ 	if (!result)
+ 	{
+ 		fprintf(stderr, _("%s: out of memory\n"), progname);
+ 		exit(1);
+ 	}
+ 	return result;
+ }
  
  /*
   * strdup() and malloc() replacements that print an error and exit
***************
*** 71,119 **** PGconn *
  GetConnection(void)
  {
  	PGconn	   *tmpconn;
- 	int			argcount = 4;	/* dbname, replication, fallback_app_name,
- 								 * password */
- 	int			i;
- 	const char **keywords;
- 	const char **values;
  	char	   *password = NULL;
  	const char *tmpparam;
! 
! 	if (dbhost)
! 		argcount++;
! 	if (dbuser)
! 		argcount++;
! 	if (dbport)
! 		argcount++;
! 
! 	keywords = pg_malloc0((argcount + 1) * sizeof(*keywords));
! 	values = pg_malloc0((argcount + 1) * sizeof(*values));
! 
! 	keywords[0] = "dbname";
! 	values[0] = "replication";
! 	keywords[1] = "replication";
! 	values[1] = "true";
! 	keywords[2] = "fallback_application_name";
! 	values[2] = progname;
! 	i = 3;
  	if (dbhost)
! 	{
! 		keywords[i] = "host";
! 		values[i] = dbhost;
! 		i++;
! 	}
  	if (dbuser)
! 	{
! 		keywords[i] = "user";
! 		values[i] = dbuser;
! 		i++;
! 	}
  	if (dbport)
! 	{
! 		keywords[i] = "port";
! 		values[i] = dbport;
! 		i++;
! 	}
  
  	while (true)
  	{
--- 89,122 ----
  GetConnection(void)
  {
  	PGconn	   *tmpconn;
  	char	   *password = NULL;
  	const char *tmpparam;
! 	char		conninfo[MAXCONNINFO];
! 	int			conn_len = 0;
! 	char		*conninfo_ptr = conninfo;
! 
! 	/*
! 	 * Connect using deliberately undocumented parameter: replication. The
! 	 * database name is ignored by the server in replication mode.
! 	 */
! 	conn_len += snprintf(conninfo_ptr, (sizeof(conninfo) - conn_len),
! 	"%s dbname=replication replication=true fallback_application_name='%s' ",
! 	connection_string ? connection_string : "", progname);
! 
! 	/*
! 	 * Extra options provided are added at the end of connection string.
! 	 */
  	if (dbhost)
! 		conn_len += snprintf((conninfo_ptr + conn_len), (sizeof(conninfo) - conn_len),
! 					"host='%s' ", dbhost);
! 
  	if (dbuser)
! 		conn_len += snprintf((conninfo_ptr + conn_len), (sizeof(conninfo) - conn_len),
! 					"user=%s ", dbuser);
! 
  	if (dbport)
! 		conn_len += snprintf((conninfo_ptr + conn_len), (sizeof(conninfo) - conn_len),
! 					"port=%s ", dbport);
  
  	while (true)
  	{
***************
*** 127,144 **** GetConnection(void)
  			 * meaning this is the call for a second session to the same
  			 * database, so just forcibly reuse that password.
  			 */
! 			keywords[argcount - 1] = "password";
! 			values[argcount - 1] = dbpassword;
  			dbgetpassword = -1; /* Don't try again if this fails */
  		}
  		else if (dbgetpassword == 1)
  		{
  			password = simple_prompt(_("Password: "), 100, false);
! 			keywords[argcount - 1] = "password";
! 			values[argcount - 1] = password;
  		}
  
! 		tmpconn = PQconnectdbParams(keywords, values, true);
  
  		/*
  		 * If there is too little memory even to allocate the PGconn object
--- 130,147 ----
  			 * meaning this is the call for a second session to the same
  			 * database, so just forcibly reuse that password.
  			 */
! 			snprintf((conninfo_ptr + conn_len), (sizeof(conninfo) - conn_len),
! 						"password='%s' ", escape_quotes(dbpassword));
  			dbgetpassword = -1; /* Don't try again if this fails */
  		}
  		else if (dbgetpassword == 1)
  		{
  			password = simple_prompt(_("Password: "), 100, false);
! 			snprintf((conninfo_ptr + conn_len), (sizeof(conninfo) - conn_len),
! 						"password='%s' ", escape_quotes(password));
  		}
  
! 		tmpconn = PQconnectdb(conninfo);
  
  		/*
  		 * If there is too little memory even to allocate the PGconn object
***************
*** 165,179 **** GetConnection(void)
  			fprintf(stderr, _("%s: could not connect to server: %s\n"),
  					progname, PQerrorMessage(tmpconn));
  			PQfinish(tmpconn);
- 			free(values);
- 			free(keywords);
  			return NULL;
  		}
  
- 		/* Connection ok! */
- 		free(values);
- 		free(keywords);
- 
  		/*
  		 * Ensure we have the same value of integer timestamps as the server
  		 * we are connecting to.
--- 168,176 ----
*** a/src/bin/pg_basebackup/streamutil.h
--- b/src/bin/pg_basebackup/streamutil.h
***************
*** 5,10 **** extern char *dbhost;
--- 5,11 ----
  extern char *dbuser;
  extern char *dbport;
  extern int	dbgetpassword;
+ extern char	*connection_string;
  
  /* Connection kept global so we can disconnect easily */
  extern PGconn *conn;
***************
*** 15,20 **** extern PGconn *conn;
--- 16,22 ----
  	exit(code);									\
  	}
  
+ extern char *escape_quotes(const char *src);
  
  extern char *pg_strdup(const char *s);
  extern void *pg_malloc0(size_t size);
*** a/src/include/pg_config_manual.h
--- b/src/include/pg_config_manual.h
***************
*** 200,205 ****
--- 200,211 ----
  #endif
  
  /*
+  * MAXCONNINFO: maximum size of a connection string.
+  */
+ #define MAXCONNINFO		1024
+ 
+ 
+ /*
   *------------------------------------------------------------------------
   * The following symbols are for enabling debugging code, not for
   * controlling user-visible features or resource limits.
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 23,35 **** extern int	wal_receiver_status_interval;
  extern int	wal_receiver_timeout;
  extern bool hot_standby_feedback;
  
- /*
-  * MAXCONNINFO: maximum size of a connection string.
-  *
-  * XXX: Should this move to pg_config_manual.h?
-  */
- #define MAXCONNINFO		1024
- 
  /* Can we allow the standby to accept replication connection from another standby? */
  #define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)
  
--- 23,28 ----
*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
***************
*** 4216,4224 **** conninfo_parse(const char *conninfo, PQExpBuffer errorMessage,
  				}
  				if (*cp == '\'')
  				{
- 					*cp2 = '\0';
  					cp++;
! 					break;
  				}
  				*cp2++ = *cp++;
  			}
--- 4216,4227 ----
  				}
  				if (*cp == '\'')
  				{
  					cp++;
! 					if (*cp != '\'')
! 					{
! 						*cp2 = '\0';
! 						break;
! 					}
  				}
  				*cp2++ = *cp++;
  			}