Fix pg_basebackup/pg_receivexlog for floating point timestamps

Since the replication protocol deals with TimestampTz, we need to
care for the floating point case as well in the frontend tools.

Fujii Masao, with changes from Magnus Hagander

doc/src/sgml/ref/pg_basebackup.sgml

@@ -338,8 +338,8 @@ PostgreSQL documentation
         Specifies the number of seconds between status packets sent back to the
         server. This is required when streaming the transaction log (using
         <literal>--xlog=stream</literal>) if replication timeout is configured
-        on the server, and allows for easier monitoring. The default value is
+        on the server, and allows for easier monitoring. A value of zero disables
-        10 seconds.
+        the status updates completely. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>

doc/src/sgml/ref/pg_receivexlog.sgml

@@ -129,8 +129,8 @@ PostgreSQL documentation
        <para>
         Specifies the number of seconds between status packets sent back to the
         server. This is required if replication timeout is configured on the
-        server, and allows for easier monitoring. The default value is
+        server, and allows for easier monitoring. A value of zero disables the
-        10 seconds.
+        status updates completely. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>

src/bin/pg_basebackup/pg_basebackup.c

@@ -46,7 +46,7 @@ int			compresslevel = 0;
 bool		includewal = false;
 bool		streamwal = false;
 bool		fastcheckpoint = false;
-int			standby_message_timeout = 10;		/* 10 sec = default */
+int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
 
 /* Progress counters */
 static uint64 totalsize;
@@ -1311,7 +1311,7 @@ main(int argc, char **argv)
 				dbgetpassword = 1;
 				break;
 			case 's':
-				standby_message_timeout = atoi(optarg);
+				standby_message_timeout = atoi(optarg) * 1000;
 				if (standby_message_timeout < 0)
 				{
 					fprintf(stderr, _("%s: invalid status interval \"%s\"\n"),

src/bin/pg_basebackup/pg_receivexlog.c

@@ -40,7 +40,7 @@
 char	   *basedir = NULL;
 int			verbose = 0;
 int			noloop = 0;
-int			standby_message_timeout = 10;		/* 10 sec = default */
+int			standby_message_timeout = 10 * 1000;		/* 10 sec = default */
 volatile bool time_to_abort = false;
 
 
@@ -356,7 +356,7 @@ main(int argc, char **argv)
 				dbgetpassword = 1;
 				break;
 			case 's':
-				standby_message_timeout = atoi(optarg);
+				standby_message_timeout = atoi(optarg) * 1000;
 				if (standby_message_timeout < 0)
 				{
 					fprintf(stderr, _("%s: invalid status interval \"%s\"\n"),

src/bin/pg_basebackup/receivelog.c

@@ -23,6 +23,7 @@
 #include "access/xlog_internal.h"
 #include "replication/walprotocol.h"
 #include "utils/datetime.h"
+#include "utils/timestamp.h"
 
 #include "receivelog.h"
 #include "streamutil.h"
@@ -195,6 +196,51 @@ localGetCurrentTimestamp(void)
 	return result;
 }
 
+/*
+ * Local version of TimestampDifference(), since we are not
+ * linked with backend code.
+ */
+static void
+localTimestampDifference(TimestampTz start_time, TimestampTz stop_time,
+					long *secs, int *microsecs)
+{
+	TimestampTz diff = stop_time - start_time;
+
+	if (diff <= 0)
+	{
+		*secs = 0;
+		*microsecs = 0;
+	}
+	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
+	}
+}
+
+/*
+ * Local version of TimestampDifferenceExceeds(), since we are not
+ * linked with backend code.
+ */
+static bool
+localTimestampDifferenceExceeds(TimestampTz start_time,
+						   TimestampTz stop_time,
+						   int msec)
+{
+	TimestampTz diff = stop_time - start_time;
+
+#ifdef HAVE_INT64_TIMESTAMP
+	return (diff >= msec * INT64CONST(1000));
+#else
+	return (diff * 1000.0 >= msec);
+#endif
+}
+
 /*
  * Receive a log stream starting at the specified position.
  *
@@ -306,7 +352,8 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
 		 */
 		now = localGetCurrentTimestamp();
 		if (standby_message_timeout > 0 &&
-			last_status < now - standby_message_timeout * 1000000)
+			localTimestampDifferenceExceeds(last_status, now,
+											standby_message_timeout))
 		{
 			/* Time to send feedback! */
 			char		replybuf[sizeof(StandbyReplyMessage) + 1];
@@ -345,10 +392,16 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
 			FD_SET(PQsocket(conn), &input_mask);
 			if (standby_message_timeout)
 			{
-				timeout.tv_sec = last_status + standby_message_timeout - now - 1;
+				TimestampTz	targettime;
+
+				targettime = TimestampTzPlusMilliseconds(last_status,
+														  standby_message_timeout - 1);
+				localTimestampDifference(now,
+										 targettime,
+										 &timeout.tv_sec,
+										 (int *)&timeout.tv_usec);
 				if (timeout.tv_sec <= 0)
 					timeout.tv_sec = 1; /* Always sleep at least 1 sec */
-				timeout.tv_usec = 0;
 				timeoutptr = &timeout;
 			}
 			else