*** a/doc/src/sgml/ref/pg_receivexlog.sgml --- b/doc/src/sgml/ref/pg_receivexlog.sgml *************** *** 106,111 **** PostgreSQL documentation --- 106,121 ---- + + + + + Allow pg_receivexlog to be used as synchronous standby. + + + + + *** a/src/backend/replication/syncrep.c --- b/src/backend/replication/syncrep.c *************** *** 376,385 **** SyncRepReleaseWaiters(void) /* * If this WALSender is serving a standby that is not on the list of * potential standbys then we have nothing to do. If we are still starting ! * up or still running base backup, then leave quickly also. */ if (MyWalSnd->sync_standby_priority == 0 || ! MyWalSnd->state < WALSNDSTATE_STREAMING) return; /* --- 376,387 ---- /* * If this WALSender is serving a standby that is not on the list of * potential standbys then we have nothing to do. If we are still starting ! * up, still running base backup or the current flush position is still ! * invalid, then leave quickly also. */ if (MyWalSnd->sync_standby_priority == 0 || ! MyWalSnd->state < WALSNDSTATE_STREAMING || ! XLogRecPtrIsInvalid(MyWalSnd->flush)) return; /* *************** *** 399,405 **** SyncRepReleaseWaiters(void) walsnd->state == WALSNDSTATE_STREAMING && walsnd->sync_standby_priority > 0 && (priority == 0 || ! priority > walsnd->sync_standby_priority)) { priority = walsnd->sync_standby_priority; syncWalSnd = walsnd; --- 401,408 ---- walsnd->state == WALSNDSTATE_STREAMING && walsnd->sync_standby_priority > 0 && (priority == 0 || ! priority > walsnd->sync_standby_priority) && ! !XLogRecPtrIsInvalid(walsnd->flush)) { priority = walsnd->sync_standby_priority; syncWalSnd = walsnd; *** a/src/backend/replication/walreceiver.c --- b/src/backend/replication/walreceiver.c *************** *** 281,286 **** WalReceiverMain(void) --- 281,291 ---- walrcv_connect(conninfo, startpoint); DisableWalRcvImmediateExit(); + /* Initialize LogstreamResult, reply_message and feedback_message */ + LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL); + MemSet(&reply_message, 0, sizeof(reply_message)); + MemSet(&feedback_message, 0, sizeof(feedback_message)); + /* Loop until end-of-streaming or error */ for (;;) { *** a/src/backend/replication/walsender.c --- b/src/backend/replication/walsender.c *************** *** 1501,1512 **** pg_stat_get_wal_senders(PG_FUNCTION_ARGS) if (walsnd->pid != 0) { ! sync_priority[i] = walsnd->sync_standby_priority; if (walsnd->state == WALSNDSTATE_STREAMING && walsnd->sync_standby_priority > 0 && (priority == 0 || ! priority > walsnd->sync_standby_priority)) { priority = walsnd->sync_standby_priority; sync_standby = i; --- 1501,1519 ---- if (walsnd->pid != 0) { ! /* ! * Treat the standby like pg_basebackup background process ! * which always returns an invalid flush location, as an ! * asynchronous standby. ! */ ! sync_priority[i] = XLogRecPtrIsInvalid(walsnd->flush) ? ! 0 : walsnd->sync_standby_priority; if (walsnd->state == WALSNDSTATE_STREAMING && walsnd->sync_standby_priority > 0 && (priority == 0 || ! priority > walsnd->sync_standby_priority) && ! !XLogRecPtrIsInvalid(walsnd->flush)) { priority = walsnd->sync_standby_priority; sync_standby = i; *** a/src/bin/pg_basebackup/pg_basebackup.c --- b/src/bin/pg_basebackup/pg_basebackup.c *************** *** 234,240 **** 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, --- 234,241 ---- { if (!ReceiveXlogStream(param->bgconn, param->startptr, param->timeline, param->sysidentifier, param->xlogdir, ! reached_end_position, standby_message_timeout, ! true, false)) /* * Any errors will already have been reported in the function process, *** a/src/bin/pg_basebackup/pg_receivexlog.c --- b/src/bin/pg_basebackup/pg_receivexlog.c *************** *** 40,45 **** --- 40,46 ---- char *basedir = NULL; int verbose = 0; int noloop = 0; + bool syncstandby = false; int standby_message_timeout = 10 * 1000; /* 10 sec = default */ volatile bool time_to_abort = false; *************** *** 59,64 **** usage(void) --- 60,66 ---- printf(_("\nOptions:\n")); printf(_(" -D, --directory=DIR receive transaction log files into this directory\n")); printf(_(" -n, --noloop do not loop on connection lost\n")); + printf(_(" -S, --syncstandby allow pg_receivexlog to be used as synchronous standby\n")); printf(_(" -v, --verbose output verbose messages\n")); printf(_(" -V, --version output version information, then exit\n")); printf(_(" -?, --help show this help, then exit\n")); *************** *** 272,279 **** StreamLog(void) timeline); ReceiveXlogStream(conn, startpos, timeline, NULL, basedir, ! stop_streaming, ! standby_message_timeout, false); PQfinish(conn); } --- 274,281 ---- timeline); ReceiveXlogStream(conn, startpos, timeline, NULL, basedir, ! stop_streaming, standby_message_timeout, ! false, syncstandby); PQfinish(conn); } *************** *** 305,310 **** main(int argc, char **argv) --- 307,313 ---- {"no-password", no_argument, NULL, 'w'}, {"password", no_argument, NULL, 'W'}, {"statusint", required_argument, NULL, 's'}, + {"syncstandby", no_argument, NULL, 'S'}, {"verbose", no_argument, NULL, 'v'}, {NULL, 0, NULL, 0} }; *************** *** 330,336 **** main(int argc, char **argv) } } ! while ((c = getopt_long(argc, argv, "D:h:p:U:s:nwWv", long_options, &option_index)) != -1) { switch (c) --- 333,339 ---- } } ! while ((c = getopt_long(argc, argv, "D:h:p:U:s:nSwWv", long_options, &option_index)) != -1) { switch (c) *************** *** 368,373 **** main(int argc, char **argv) --- 371,379 ---- exit(1); } break; + case 'S': + syncstandby = true; + break; case 'n': noloop = 1; break; *** a/src/bin/pg_basebackup/receivelog.c --- b/src/bin/pg_basebackup/receivelog.c *************** *** 264,270 **** localTimestampDifferenceExceeds(TimestampTz start_time, * Note: The log position *must* be at a log segment start! */ 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]; --- 264,274 ---- * Note: The log position *must* be at a log segment start! */ bool ! ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, ! char *sysidentifier, char *basedir, ! stream_stop_callback stream_stop, ! int standby_message_timeout, ! bool rename_partial, bool syncstandby) { char query[128]; char current_walfile_name[MAXPGPATH]; *************** *** 273,278 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi --- 277,285 ---- int walfile = -1; int64 last_status = -1; XLogRecPtr blockpos = InvalidXLogRecPtr; + XLogRecPtr flushpos = InvalidXLogRecPtr; + char replybuf[sizeof(StandbyReplyMessage) + 1]; + StandbyReplyMessage *replymsg = (StandbyReplyMessage *) (replybuf + 1); if (sysidentifier != NULL) { *************** *** 351,366 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi * Potentially send a status message to the master */ now = localGetCurrentTimestamp(); ! if (standby_message_timeout > 0 && ! localTimestampDifferenceExceeds(last_status, now, ! standby_message_timeout)) { /* Time to send feedback! */ ! char replybuf[sizeof(StandbyReplyMessage) + 1]; ! StandbyReplyMessage *replymsg = (StandbyReplyMessage *) (replybuf + 1); ! ! replymsg->write = InvalidXLogRecPtr; ! replymsg->flush = InvalidXLogRecPtr; replymsg->apply = InvalidXLogRecPtr; replymsg->sendTime = now; replybuf[0] = 'r'; --- 358,373 ---- * Potentially send a status message to the master */ now = localGetCurrentTimestamp(); ! if ((standby_message_timeout > 0 && ! localTimestampDifferenceExceeds(last_status, now, ! standby_message_timeout)) || ! (syncstandby && ! (!XLByteEQ(replymsg->write, blockpos) || ! !XLByteEQ(replymsg->flush, flushpos)))) { /* Time to send feedback! */ ! replymsg->write = blockpos; ! replymsg->flush = syncstandby ? flushpos : InvalidXLogRecPtr; replymsg->apply = InvalidXLogRecPtr; replymsg->sendTime = now; replybuf[0] = 'r'; *************** *** 550,555 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi --- 557,563 ---- walfile = -1; xlogoff = 0; + flushpos = blockpos; if (stream_stop != NULL) { *** a/src/bin/pg_basebackup/receivelog.h --- b/src/bin/pg_basebackup/receivelog.h *************** *** 13,16 **** extern bool ReceiveXlogStream(PGconn *conn, char *basedir, stream_stop_callback stream_stop, int standby_message_timeout, ! bool rename_partial); --- 13,17 ---- char *basedir, stream_stop_callback stream_stop, int standby_message_timeout, ! bool rename_partial, ! bool syncstandby);