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

diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml
index 76bfcf1cbb854145333e5757373ef942613b8a83..102d649f4e8b52cff30699bbb6db8641b8302cd4 100644 (file)
--- 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
         --xlog=stream) 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.
        
       
      

diff --git a/doc/src/sgml/ref/pg_receivexlog.sgml b/doc/src/sgml/ref/pg_receivexlog.sgml
index f829170da2ce5d45f7744a667c658aa242eec8c1..b1eee1f20b822ee0a5d2067d646a1fb7fda6b45e 100644 (file)
--- a/doc/src/sgml/ref/pg_receivexlog.sgml
+++ b/doc/src/sgml/ref/pg_receivexlog.sgml
@@ -129,8 +129,8 @@ PostgreSQL documentation
        
         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.
        
       
      

diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index d7466168d7856afd336de818416c7186830368ed..c3a0d89897ad553d7d28bb12ea20d431c3c46d34 100644 (file)
--- 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
index 084ddc4a8cb001b8ab328443fbcc40746cc55922..67a70bcf713e4b7b1c618fde9035705851df6df7 100644 (file)
--- 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
index 663977185238c5b4d6efc41ffe3769506b60e5f3..a51a40edfd17fb892e5faefef5d94b63684dc600 100644 (file)
--- 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