From a0b4c5a20a5e5076225ba440a921f4b46f1159ee Mon Sep 17 00:00:00 2001 From: Magnus Hagander Date: Sun, 10 Jun 2012 12:12:36 +0200 Subject: [PATCH] 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 | 4 +- doc/src/sgml/ref/pg_receivexlog.sgml | 4 +- src/bin/pg_basebackup/pg_basebackup.c | 4 +- src/bin/pg_basebackup/pg_receivexlog.c | 4 +- src/bin/pg_basebackup/receivelog.c | 59 ++++++++++++++++++++++++-- 5 files changed, 64 insertions(+), 11 deletions(-) diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml index 76bfcf1cbb..102d649f4e 100644 --- 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 f829170da2..b1eee1f20b 100644 --- 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 d7466168d7..c3a0d89897 100644 --- 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 084ddc4a8c..67a70bcf71 100644 --- 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 6639771852..a51a40edfd 100644 --- 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