Fix pg_basebackup/pg_receivexlog for floating point timestamps

mhagander · mhagander · commit a0b4c5a20a5e · 2012-06-10T12:12:36.000+02:00
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
diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/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
-        10 seconds.
+        on the server, and allows for easier monitoring. A value of zero disables
+        the status updates completely. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>
diff --git a/doc/src/sgml/ref/pg_receivexlog.sgml b/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
-        10 seconds.
+        server, and allows for easier monitoring. A value of zero disables the
+        status updates completely. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/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"),
diff --git a/src/bin/pg_basebackup/pg_receivexlog.c b/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"),
diff --git a/src/bin/pg_basebackup/receivelog.c b/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
