*** a/doc/src/sgml/config.sgml --- b/doc/src/sgml/config.sgml *************** *** 2015,2020 **** SET ENABLE_SEQSCAN TO OFF; --- 2015,2044 ---- + + + replication_timeout (integer) + + replication_timeout configuration parameter + + + + Specifies the maximum time, in milliseconds, to wait for the reply + from the standby before terminating replication. This is useful for + the primary server to detect the standby crash or network outage. + A value of zero (the default) turns this off. This parameter can + only be set in the postgresql.conf file or on the server + command line. + + + To make the timeout work properly, + must be enabled on the standby, and its value must be less than the + value of replication_timeout. + If wal_receiver_status_interval is zero on the standby, + replication timeout is disabled for that connection. + + + *************** *** 2211,2216 **** SET ENABLE_SEQSCAN TO OFF; --- 2235,2245 ---- the postgresql.conf file or on the server command line. The default value is 10 seconds. + + When is enabled on the primary, + wal_receiver_status_interval must be enabled, and its value + must be less than the value of replication_timeout. + *** a/doc/src/sgml/protocol.sgml --- b/doc/src/sgml/protocol.sgml *************** *** 1480,1485 **** The commands accepted in walsender mode are: --- 1480,1533 ---- + Standby parameter change (F) + + + + + + + Byte1('g') + + + + Identifies the message as a receiver parameter change. + + + + + + Int32 + + + + The current value of wal_receiver_status_interval + in the standby. + + + + + + Byte8 + + + + The server's system clock at the time of transmission, + given in TimestampTz format. + + + + + + + + + + + + + + Standby status update (F) *** a/src/backend/libpq/pqcomm.c --- b/src/backend/libpq/pqcomm.c *************** *** 56,61 **** --- 56,64 ---- * pq_putbytes - send bytes to connection (not flushed until pq_flush) * pq_flush - flush pending output * pq_getbyte_if_available - get a byte if available without blocking + * pq_putbytes_if_writable - send bytes to connection if writable without blocking + * pq_flush_if_writable - flush pending output if writable without blocking + * pq_set_nonblocking - set socket blocking/non-blocking * * message-level I/O (and old-style-COPY-OUT cruft): * pq_putmessage - send a normal message (suppressed in COPY OUT mode) *************** *** 112,117 **** static char sock_path[MAXPGPATH]; --- 115,121 ---- static char PqSendBuffer[PQ_BUFFER_SIZE]; static int PqSendPointer; /* Next index to store a byte in PqSendBuffer */ + static int PqSendStart; /* Next index to send a byte in PqSendBuffer */ static char PqRecvBuffer[PQ_BUFFER_SIZE]; static int PqRecvPointer; /* Next index to read a byte from PqRecvBuffer */ *************** *** 127,133 **** static bool DoingCopyOut; /* Internal functions */ static void pq_close(int code, Datum arg); static int internal_putbytes(const char *s, size_t len); ! static int internal_flush(void); #ifdef HAVE_UNIX_SOCKETS static int Lock_AF_UNIX(unsigned short portNumber, char *unixSocketName); --- 131,138 ---- /* Internal functions */ static void pq_close(int code, Datum arg); static int internal_putbytes(const char *s, size_t len); ! static int internal_flush(bool nonblocking); ! static void pq_set_nonblocking(bool nonblocking, int emode); #ifdef HAVE_UNIX_SOCKETS static int Lock_AF_UNIX(unsigned short portNumber, char *unixSocketName); *************** *** 142,148 **** static int Setup_AF_UNIX(void); void pq_init(void) { ! PqSendPointer = PqRecvPointer = PqRecvLength = 0; PqCommBusy = false; DoingCopyOut = false; on_proc_exit(pq_close, 0); --- 147,153 ---- void pq_init(void) { ! PqSendPointer = PqSendStart = PqRecvPointer = PqRecvLength = 0; PqCommBusy = false; DoingCopyOut = false; on_proc_exit(pq_close, 0); *************** *** 846,859 **** pq_getbyte_if_available(unsigned char *c) } /* Temporarily put the socket into non-blocking mode */ ! #ifdef WIN32 ! pgwin32_noblock = 1; ! #else ! if (!pg_set_noblock(MyProcPort->sock)) ! ereport(ERROR, ! (errmsg("could not set socket to non-blocking mode: %m"))); ! #endif ! MyProcPort->noblock = true; PG_TRY(); { r = secure_read(MyProcPort, c, 1); --- 851,857 ---- } /* Temporarily put the socket into non-blocking mode */ ! pq_set_nonblocking(true, ERROR); PG_TRY(); { r = secure_read(MyProcPort, c, 1); *************** *** 892,916 **** pq_getbyte_if_available(unsigned char *c) * The rest of the backend code assumes the socket is in blocking * mode, so treat failure as FATAL. */ ! #ifdef WIN32 ! pgwin32_noblock = 0; ! #else ! if (!pg_set_block(MyProcPort->sock)) ! ereport(FATAL, ! (errmsg("could not set socket to blocking mode: %m"))); ! #endif ! MyProcPort->noblock = false; PG_RE_THROW(); } PG_END_TRY(); ! #ifdef WIN32 ! pgwin32_noblock = 0; ! #else ! if (!pg_set_block(MyProcPort->sock)) ! ereport(FATAL, ! (errmsg("could not set socket to blocking mode: %m"))); ! #endif ! MyProcPort->noblock = false; return r; } --- 890,900 ---- * The rest of the backend code assumes the socket is in blocking * mode, so treat failure as FATAL. */ ! pq_set_nonblocking(false, FATAL); PG_RE_THROW(); } PG_END_TRY(); ! pq_set_nonblocking(false, FATAL); return r; } *************** *** 1139,1145 **** internal_putbytes(const char *s, size_t len) { /* If buffer is full, then flush it out */ if (PqSendPointer >= PQ_BUFFER_SIZE) ! if (internal_flush()) return EOF; amount = PQ_BUFFER_SIZE - PqSendPointer; if (amount > len) --- 1123,1129 ---- { /* If buffer is full, then flush it out */ if (PqSendPointer >= PQ_BUFFER_SIZE) ! if (internal_flush(false) == EOF) return EOF; amount = PQ_BUFFER_SIZE - PqSendPointer; if (amount > len) *************** *** 1153,1158 **** internal_putbytes(const char *s, size_t len) --- 1137,1192 ---- } /* -------------------------------- + * pq_putbytes_if_writable - send bytes to connection (not flushed + * until pq_flush), if writable + * + * Returns the number of bytes written without blocking, or EOF if trouble. + * -------------------------------- + */ + int + pq_putbytes_if_writable(const char *s, size_t len) + { + size_t amount; + size_t nwritten = 0; + + /* Should not be called by old-style COPY OUT */ + Assert(!DoingCopyOut); + /* No-op if reentrant call */ + if (PqCommBusy) + return 0; + PqCommBusy = true; + + while (len > 0) + { + /* If buffer is full, then flush it out */ + if (PqSendPointer >= PQ_BUFFER_SIZE) + { + int r; + + r = internal_flush(true); + if (r == 0) + break; + if (r == EOF) + { + PqCommBusy = false; + return r; + } + } + amount = PQ_BUFFER_SIZE - PqSendPointer; + if (amount > len) + amount = len; + memcpy(PqSendBuffer + PqSendPointer, s, amount); + PqSendPointer += amount; + s += amount; + len -= amount; + nwritten += amount; + } + + PqCommBusy = false; + return (int) nwritten; + } + + /* -------------------------------- * pq_flush - flush pending output * * returns 0 if OK, EOF if trouble *************** *** 1167,1227 **** pq_flush(void) if (PqCommBusy) return 0; PqCommBusy = true; ! res = internal_flush(); PqCommBusy = false; ! return res; } static int ! internal_flush(void) { static int last_reported_send_errno = 0; ! char *bufptr = PqSendBuffer; char *bufend = PqSendBuffer + PqSendPointer; ! while (bufptr < bufend) { ! int r; ! ! r = secure_write(MyProcPort, bufptr, bufend - bufptr); ! ! if (r <= 0) { ! if (errno == EINTR) ! continue; /* Ok if we were interrupted */ ! /* ! * Careful: an ereport() that tries to write to the client would ! * cause recursion to here, leading to stack overflow and core ! * dump! This message must go *only* to the postmaster log. ! * ! * If a client disconnects while we're in the midst of output, we ! * might write quite a bit of data before we get to a safe query ! * abort point. So, suppress duplicate log messages. ! */ ! if (errno != last_reported_send_errno) { ! last_reported_send_errno = errno; ! ereport(COMMERROR, ! (errcode_for_socket_access(), ! errmsg("could not send data to client: %m"))); } ! /* ! * We drop the buffered data anyway so that processing can ! * continue, even though we'll probably quit soon. ! */ ! PqSendPointer = 0; ! return EOF; } - - last_reported_send_errno = 0; /* reset after any successful send */ - bufptr += r; } ! PqSendPointer = 0; ! return 0; } --- 1201,1362 ---- if (PqCommBusy) return 0; PqCommBusy = true; ! res = internal_flush(false); PqCommBusy = false; ! return (res == 1) ? 0 : EOF; } + /* -------------------------------- + * internal_flush - flush pending output + * + * Returns 1 if OK, 0 if pending output cannot be written without blocking + * (only possible nonblocking is true), or EOF if trouble. + * -------------------------------- + */ static int ! internal_flush(bool nonblocking) { static int last_reported_send_errno = 0; + int r; ! char *bufptr = PqSendBuffer + PqSendStart; char *bufend = PqSendBuffer + PqSendPointer; ! /* Temporarily put the socket into non-blocking mode */ ! if (nonblocking) ! pq_set_nonblocking(true, ERROR); ! PG_TRY(); { ! while (bufptr < bufend) { ! r = secure_write(MyProcPort, bufptr, bufend - bufptr); ! if (r <= 0) { ! /* Ok if we were interrupted in blocking mode */ ! if (!nonblocking && errno == EINTR) ! continue; ! ! if (nonblocking) ! { ! if (r == 0) ! r = EOF; /* EOF detected */ ! else if (errno == EAGAIN || ! errno == EWOULDBLOCK || ! errno == EINTR) ! { ! /* ! * Ok if no data writable without blocking or ! * interrupted (though EINTR really shouldn't ! * happen with a non-blocking socket). Report ! * other errors. ! */ ! r = 0; ! } ! break; ! } ! ! /* ! * Careful: an ereport() that tries to write to the ! * client would cause recursion to here, leading to ! * stack overflow and core dump! This message must ! * go *only* to the postmaster log. ! * ! * If a client disconnects while we're in the midst ! * of output, we might write quite a bit of data before ! * we get to a safe query abort point. So, suppress ! * duplicate log messages. ! */ ! if (errno != last_reported_send_errno) ! { ! last_reported_send_errno = errno; ! ereport(COMMERROR, ! (errcode_for_socket_access(), ! errmsg("could not send data to client: %m"))); ! } ! ! /* ! * We drop the buffered data anyway so that processing can ! * continue, even though we'll probably quit soon. ! */ ! PqSendStart = PqSendPointer = 0; ! r = EOF; ! break; } ! last_reported_send_errno = 0; /* reset after any successful send */ ! bufptr += r; ! PqSendStart += r; } } + PG_CATCH(); + { + /* + * The rest of the backend code assumes the socket is in blocking + * mode, so treat failure as FATAL. + */ + if (nonblocking) + pq_set_nonblocking(false, FATAL); + } + PG_END_TRY(); + if (nonblocking) + pq_set_nonblocking(false, FATAL); ! if (r == 0 || r == EOF) ! return r; ! ! PqSendStart = PqSendPointer = 0; ! return 1; ! } ! ! /* -------------------------------- ! * pq_flush_if_writable - flush pending output if writable ! * ! * Returns 1 if OK, 0 if pending output cannot be written without blocking, ! * or EOF if trouble. ! * -------------------------------- ! */ ! int ! pq_flush_if_writable(void) ! { ! int res; ! ! /* No-op if reentrant call */ ! if (PqCommBusy) ! return 0; ! PqCommBusy = true; ! res = internal_flush(true); ! PqCommBusy = false; ! return res; ! } ! ! /* -------------------------------- ! * pq_set_nonblocking - set socket blocking/non-blocking ! * ! * Sets the socket non-blocking if nonblocking is TRUE, or sets it ! * blocking otherwise. ! * -------------------------------- ! */ ! static ! void pq_set_nonblocking(bool nonblocking, int emode) ! { ! #ifdef WIN32 ! pgwin32_noblock = nonblocking ? 1 : 0; ! #else ! if (nonblocking) ! { ! if (!pg_set_noblock(MyProcPort->sock)) ! ereport(emode, ! (errmsg("could not set socket to non-blocking mode: %m"))); ! } ! else ! { ! if (!pg_set_block(MyProcPort->sock)) ! ereport(emode, ! (errmsg("could not set socket to blocking mode: %m"))); ! } ! #endif ! MyProcPort->noblock = nonblocking; } *** a/src/backend/port/unix_latch.c --- b/src/backend/port/unix_latch.c *************** *** 193,211 **** DisownLatch(volatile Latch *latch) bool WaitLatch(volatile Latch *latch, long timeout) { ! return WaitLatchOrSocket(latch, PGINVALID_SOCKET, timeout) > 0; } /* * Like WaitLatch, but will also return when there's data available in ! * 'sock' for reading. Returns 0 if timeout was reached, 1 if the latch ! * was set, or 2 if the scoket became readable. */ int ! WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout) { struct timeval tv, *tvp = NULL; fd_set input_mask; int rc; int result = 0; --- 193,214 ---- bool WaitLatch(volatile Latch *latch, long timeout) { ! return WaitLatchOrSocket(latch, PGINVALID_SOCKET, false, false, timeout) > 0; } /* * Like WaitLatch, but will also return when there's data available in ! * 'sock' for reading or writing. Returns 0 if timeout was reached, ! * 1 if the latch was set, 2 if the scoket became readable, or 3 if ! * the socket became writable. */ int ! WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, bool forRead, ! bool forWrite, long timeout) { struct timeval tv, *tvp = NULL; fd_set input_mask; + fd_set output_mask; int rc; int result = 0; *************** *** 241,254 **** WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout) FD_ZERO(&input_mask); FD_SET(selfpipe_readfd, &input_mask); hifd = selfpipe_readfd; ! if (sock != PGINVALID_SOCKET) { FD_SET(sock, &input_mask); if (sock > hifd) hifd = sock; } ! rc = select(hifd + 1, &input_mask, NULL, NULL, tvp); if (rc < 0) { if (errno == EINTR) --- 244,265 ---- FD_ZERO(&input_mask); FD_SET(selfpipe_readfd, &input_mask); hifd = selfpipe_readfd; ! if (sock != PGINVALID_SOCKET && forRead) { FD_SET(sock, &input_mask); if (sock > hifd) hifd = sock; } ! FD_ZERO(&output_mask); ! if (sock != PGINVALID_SOCKET && forWrite) ! { ! FD_SET(sock, &output_mask); ! if (sock > hifd) ! hifd = sock; ! } ! ! rc = select(hifd + 1, &input_mask, &output_mask, NULL, tvp); if (rc < 0) { if (errno == EINTR) *************** *** 263,273 **** WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout) result = 0; break; } ! if (sock != PGINVALID_SOCKET && FD_ISSET(sock, &input_mask)) { result = 2; break; /* data available in socket */ } } waiting = false; --- 274,291 ---- result = 0; break; } ! if (sock != PGINVALID_SOCKET && forRead && ! FD_ISSET(sock, &input_mask)) { result = 2; break; /* data available in socket */ } + if (sock != PGINVALID_SOCKET && forWrite && + FD_ISSET(sock, &output_mask)) + { + result = 3; + break; /* data writable in socket */ + } } waiting = false; *** a/src/backend/port/win32/socket.c --- b/src/backend/port/win32/socket.c *************** *** 14,20 **** #include "postgres.h" /* ! * Indicate if pgwin32_recv() should operate in non-blocking mode. * * Since the socket emulation layer always sets the actual socket to * non-blocking mode in order to be able to deliver signals, we must --- 14,21 ---- #include "postgres.h" /* ! * Indicate if pgwin32_recv() and pgwin32_send() should operate ! * in non-blocking mode. * * Since the socket emulation layer always sets the actual socket to * non-blocking mode in order to be able to deliver signals, we must *************** *** 399,404 **** pgwin32_send(SOCKET s, char *buf, int len, int flags) --- 400,415 ---- return -1; } + if (pgwin32_noblock) + { + /* + * No data sent, and we are in "emulated non-blocking mode", so + * return indicating that we'd block if we were to continue. + */ + errno = EWOULDBLOCK; + return -1; + } + /* No error, zero bytes (win2000+) or error+WSAEWOULDBLOCK (<=nt4) */ if (pgwin32_waitforsinglesocket(s, FD_WRITE | FD_CLOSE, INFINITE) == 0) *** a/src/backend/port/win32_latch.c --- b/src/backend/port/win32_latch.c *************** *** 85,95 **** DisownLatch(volatile Latch *latch) bool WaitLatch(volatile Latch *latch, long timeout) { ! return WaitLatchOrSocket(latch, PGINVALID_SOCKET, timeout) > 0; } int ! WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout) { DWORD rc; HANDLE events[3]; --- 85,96 ---- bool WaitLatch(volatile Latch *latch, long timeout) { ! return WaitLatchOrSocket(latch, PGINVALID_SOCKET, false, false, timeout) > 0; } int ! WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, bool forRead, ! bool forWrite, long timeout) { DWORD rc; HANDLE events[3]; *************** *** 103,112 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout) events[0] = latchevent; events[1] = pgwin32_signal_event; numevents = 2; ! if (sock != PGINVALID_SOCKET) { sockevent = WSACreateEvent(); ! WSAEventSelect(sock, sockevent, FD_READ); events[numevents++] = sockevent; } --- 104,120 ---- events[0] = latchevent; events[1] = pgwin32_signal_event; numevents = 2; ! if (sock != PGINVALID_SOCKET && (forRead || forWrite)) { + int flags = 0; + + if (forRead) + flags |= FD_READ; + if (forWrite) + flags |= FD_WRITE; + sockevent = WSACreateEvent(); ! WSAEventSelect(sock, sockevent, flags); events[numevents++] = sockevent; } *************** *** 139,146 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout) pgwin32_dispatch_queued_signals(); else if (rc == WAIT_OBJECT_0 + 2) { Assert(sock != PGINVALID_SOCKET); ! result = 2; break; } else if (rc != WAIT_OBJECT_0) --- 147,165 ---- pgwin32_dispatch_queued_signals(); else if (rc == WAIT_OBJECT_0 + 2) { + WSANETWORKEVENTS resEvents; + Assert(sock != PGINVALID_SOCKET); ! ! ZeroMemory(&resEvents, sizeof(resEvents)); ! if (WSAEnumNetworkEvents(sock, sockevent, &resEvents) == SOCKET_ERROR) ! ereport(FATAL, ! (errmsg_internal("failed to enumerate network events: %i", (int) GetLastError()))); ! ! if (forRead && resEvents.lNetworkEvents & FD_READ) ! result = 2; ! if (forWrite && resEvents.lNetworkEvents & FD_WRITE) ! result = 3; break; } else if (rc != WAIT_OBJECT_0) *************** *** 148,154 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout) } /* Clean up the handle we created for the socket */ ! if (sock != PGINVALID_SOCKET) { WSAEventSelect(sock, sockevent, 0); WSACloseEvent(sockevent); --- 167,173 ---- } /* Clean up the handle we created for the socket */ ! if (sock != PGINVALID_SOCKET && (forRead || forWrite)) { WSAEventSelect(sock, sockevent, 0); WSACloseEvent(sockevent); *** a/src/backend/replication/walreceiver.c --- b/src/backend/replication/walreceiver.c *************** *** 125,130 **** static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr); --- 125,131 ---- static void XLogWalRcvFlush(bool dying); static void XLogWalRcvSendReply(void); static void XLogWalRcvSendHSFeedback(void); + static void XLogWalRcvSendGUCChange(void); /* Signal handlers */ static void WalRcvSigHupHandler(SIGNAL_ARGS); *************** *** 276,281 **** WalReceiverMain(void) --- 277,288 ---- walrcv_connect(conninfo, startpoint); DisableWalRcvImmediateExit(); + /* + * Report the important parameters for streaming replication to + * the primary. + */ + XLogWalRcvSendGUCChange(); + /* Loop until end-of-streaming or error */ for (;;) { *************** *** 303,310 **** WalReceiverMain(void) --- 310,325 ---- if (got_SIGHUP) { + int save_wal_receiver_status_interval = + wal_receiver_status_interval; + got_SIGHUP = false; ProcessConfigFile(PGC_SIGHUP); + + /* If any of important parameters have changed, report them */ + if (save_wal_receiver_status_interval != + wal_receiver_status_interval) + XLogWalRcvSendGUCChange(); } /* Wait a while for data to arrive */ *************** *** 704,706 **** XLogWalRcvSendHSFeedback(void) --- 719,748 ---- memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage)); walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1); } + + /* + * Send parameter change message to primary, plus the current time. + */ + static void + XLogWalRcvSendGUCChange(void) + { + char buf[sizeof(StandbyGUCChangeMessage) + 1]; + TimestampTz now; + StandbyGUCChangeMessage guc_change_message; + + /* Get current timestamp. */ + now = GetCurrentTimestamp(); + + /* Construct a new message */ + guc_change_message.wal_receiver_status_interval = + wal_receiver_status_interval; + guc_change_message.sendTime = now; + + elog(DEBUG2, "sending parameter change wal_receiver_status_interval %d", + wal_receiver_status_interval); + + /* Prepend with the message type and send it. */ + buf[0] = 'g'; + memcpy(&buf[1], &guc_change_message, sizeof(StandbyGUCChangeMessage)); + walrcv_send(buf, sizeof(StandbyGUCChangeMessage) + 1); + } *** a/src/backend/replication/walsender.c --- b/src/backend/replication/walsender.c *************** *** 74,79 **** bool am_walsender = false; /* Am I a walsender process ? */ --- 74,91 ---- /* User-settable parameters for walsender */ int max_wal_senders = 0; /* the maximum number of concurrent walsenders */ int WalSndDelay = 1000; /* max sleep time between some actions */ + int replication_timeout = 0; /* maximum time to send one WAL data message */ + + /* + * Buffer for WAL sending + * + * WalSndOutBuffer is a work area in which the output message is constructed. + * It's used in just so we can avoid re-palloc'ing the buffer on each cycle. + * It must be of size 6 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE. + */ + static char *WalSndOutBuffer; + static int WalSndOutHead; /* head of pending output */ + static int WalSndOutTail; /* tail of pending output */ /* * These variables are used similarly to openLogFile/Id/Seg/Off, *************** *** 95,100 **** static XLogRecPtr sentPtr = {0, 0}; --- 107,123 ---- */ static StringInfoData reply_message; + /* + * Timestamp of the last receipt of the reply from the standby. + */ + static TimestampTz last_reply_timestamp; + + /* + * The value of wal_receiver_status_interval on the standby. + * If this is zero, we disable replication timeout. + */ + static int standby_wal_receiver_status_interval = 10; + /* Flags set by signal handlers for later service in main loop */ static volatile sig_atomic_t got_SIGHUP = false; volatile sig_atomic_t walsender_shutdown_requested = false; *************** *** 113,124 **** static int WalSndLoop(void); static void InitWalSnd(void); static void WalSndHandshake(void); static void WalSndKill(int code, Datum arg); ! static bool XLogSend(char *msgbuf, bool *caughtup); static void IdentifySystem(void); static void StartReplication(StartReplicationCmd * cmd); static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); --- 136,149 ---- static void InitWalSnd(void); static void WalSndHandshake(void); static void WalSndKill(int code, Datum arg); ! static bool XLogSend(bool *caughtup, bool *pending); static void IdentifySystem(void); static void StartReplication(StartReplicationCmd * cmd); static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); + static void ProcessStandbyGUCChangeMessage(void); + static void ValidateReplicationTimeout(bool standby); static void ProcessRepliesIfAny(void); *************** *** 216,221 **** WalSndHandshake(void) --- 241,252 ---- { got_SIGHUP = false; ProcessConfigFile(PGC_SIGHUP); + + /* + * Don't need to verify replication_timeout here because we + * don't receive any parameter change message until we enter + * streaming mode. + */ } if (firstchar != EOF) *************** *** 469,474 **** ProcessRepliesIfAny(void) --- 500,506 ---- { unsigned char firstchar; int r; + int received = false; for (;;) { *************** *** 481,489 **** ProcessRepliesIfAny(void) errmsg("unexpected EOF on standby connection"))); proc_exit(0); } ! if (r == 0) { ! /* no data available without blocking */ return; } --- 513,526 ---- errmsg("unexpected EOF on standby connection"))); proc_exit(0); } ! if (r == 0) /* no data available without blocking */ { ! /* ! * Save the last reply timestamp if we've received at least ! * one reply. ! */ ! if (received) ! last_reply_timestamp = GetCurrentTimestamp(); return; } *************** *** 495,500 **** ProcessRepliesIfAny(void) --- 532,538 ---- */ case 'd': ProcessStandbyMessage(); + received = true; break; /* *************** *** 513,519 **** ProcessRepliesIfAny(void) } /* ! * Process a status update message received from standby. */ static void ProcessStandbyMessage(void) --- 551,557 ---- } /* ! * Process a message received from standby. */ static void ProcessStandbyMessage(void) *************** *** 549,554 **** ProcessStandbyMessage(void) --- 587,596 ---- ProcessStandbyHSFeedbackMessage(); break; + case 'g': + ProcessStandbyGUCChangeMessage(); + break; + default: ereport(COMMERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), *************** *** 669,687 **** ProcessStandbyHSFeedbackMessage(void) } } /* Main loop of walsender process */ static int WalSndLoop(void) { - char *output_message; bool caughtup = false; /* * Allocate buffer that will be used for each output message. We do this * just once to reduce palloc overhead. The buffer must be made large * enough for maximum-sized messages. */ ! output_message = palloc(1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE); /* * Allocate buffer that will be used for processing reply messages. As --- 711,809 ---- } } + /* + * Parameter change report from standby + */ + static void + ProcessStandbyGUCChangeMessage(void) + { + StandbyGUCChangeMessage reply; + + pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyGUCChangeMessage)); + + elog(DEBUG2, "standby's wal_receiver_status_interval %d", + reply.wal_receiver_status_interval); + + /* If we don't use replication timeout, we don't need to do any more */ + if (replication_timeout <= 0) + return; + + /* + * If wal_receiver_status_interval is zero in the standby, + * we ignore replication timeout for that connection. + */ + if (standby_wal_receiver_status_interval > 0 && + reply.wal_receiver_status_interval <= 0) + { + ereport(LOG, + (errmsg("replication timeout is disabled because wal_receiver_status_interval is zero on the standby"))); + } + else if (standby_wal_receiver_status_interval <= 0 && + reply.wal_receiver_status_interval > 0) + { + ereport(LOG, + (errmsg("replication timeout is enabled because wal_receiver_status_interval is greater than zero on the standby"))); + } + + standby_wal_receiver_status_interval = reply.wal_receiver_status_interval; + + /* Verify whether replication_timeout is large enough */ + ValidateReplicationTimeout(true); + } + + /* + * Verify whether replication_timeout is large enough for the timeout. + * + * This is called when either replication_timeout is changed on the + * master or wal_receiver_status_interval is changed on the standby. + * In the former case, 'standby' must be set to FALSE, otherwise TRUE. + */ + static void + ValidateReplicationTimeout(bool standby) + { + static bool skip_validation = true; + + /* + * If we've not received any parameter change message from the standby + * yet, we cannot verify replication_timeout properly. In that case, + * we postpone the validation until the message has arrived. This can + * happen only when replication_timeout is changed by SIGHUP signal + * before any parameter change message arrives. + */ + if (standby) + skip_validation = false; + + if (skip_validation) + return; + + /* + * Emit WARNING message if replication_timeout in the primary + * is less than wal_receiver_status_interval in the standby + * because unexpected timeout can happen in that case. + */ + if (replication_timeout > 0 && + standby_wal_receiver_status_interval > 0 && + replication_timeout <= standby_wal_receiver_status_interval * 1000) + ereport(WARNING, + (errmsg("replication can be terminated unexpectedly because replication_timeout (%d milliseconds) on the master is less than wal_receiver_status_interval (%d seconds) on the standby", + replication_timeout, standby_wal_receiver_status_interval), + errhint("Either increase replication_timeout on the master, or decrease wal_receiver_status_interval on the standby."))); + } + /* Main loop of walsender process */ static int WalSndLoop(void) { bool caughtup = false; + bool pending = false; /* * Allocate buffer that will be used for each output message. We do this * just once to reduce palloc overhead. The buffer must be made large * enough for maximum-sized messages. */ ! WalSndOutBuffer = palloc(6 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE); ! WalSndOutHead = WalSndOutTail = 0; /* * Allocate buffer that will be used for processing reply messages. As *************** *** 689,694 **** WalSndLoop(void) --- 811,819 ---- */ initStringInfo(&reply_message); + /* Initialize the last reply timestamp */ + last_reply_timestamp = GetCurrentTimestamp(); + /* Loop forever, unless we get an error */ for (;;) { *************** *** 705,710 **** WalSndLoop(void) --- 830,838 ---- got_SIGHUP = false; ProcessConfigFile(PGC_SIGHUP); SyncRepInitConfig(); + + /* Verify whether replication_timeout is large enough */ + ValidateReplicationTimeout(false); } /* *************** *** 713,722 **** WalSndLoop(void) */ if (walsender_ready_to_stop) { ! if (!XLogSend(output_message, &caughtup)) break; ProcessRepliesIfAny(); ! if (caughtup) walsender_shutdown_requested = true; } --- 841,850 ---- */ if (walsender_ready_to_stop) { ! if (!XLogSend(&caughtup, &pending)) break; ProcessRepliesIfAny(); ! if (caughtup && !pending) walsender_shutdown_requested = true; } *************** *** 731,740 **** WalSndLoop(void) } /* ! * If we had sent all accumulated WAL in last round, nap for the ! * configured time before retrying. */ ! if (caughtup) { /* * Even if we wrote all the WAL that was available when we started --- 859,869 ---- } /* ! * If we had sent all accumulated WAL in last round or could not ! * flush pending WAL in output buffer because the socket was not ! * writable, nap for the configured time before retrying. */ ! if (caughtup || pending) { /* * Even if we wrote all the WAL that was available when we started *************** *** 745,769 **** WalSndLoop(void) */ ResetLatch(&MyWalSnd->latch); ! if (!XLogSend(output_message, &caughtup)) break; ! if (caughtup && !got_SIGHUP && !walsender_ready_to_stop && !walsender_shutdown_requested) { /* * XXX: We don't really need the periodic wakeups anymore, * WaitLatchOrSocket should reliably wake up as soon as * something interesting happens. */ /* Sleep */ WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock, ! WalSndDelay * 1000L); } } else { /* Attempt to send the log once every loop */ ! if (!XLogSend(output_message, &caughtup)) break; } --- 874,936 ---- */ ResetLatch(&MyWalSnd->latch); ! if (!XLogSend(&caughtup, &pending)) break; ! if ((caughtup || pending) && !got_SIGHUP && !walsender_ready_to_stop && ! !walsender_shutdown_requested) { + TimestampTz finish_time; + long sleeptime; + /* * XXX: We don't really need the periodic wakeups anymore, * WaitLatchOrSocket should reliably wake up as soon as * something interesting happens. */ + /* Reschedule replication timeout */ + if (replication_timeout > 0 && + standby_wal_receiver_status_interval > 0) + { + long secs; + int usecs; + + finish_time = TimestampTzPlusMilliseconds(last_reply_timestamp, + replication_timeout); + TimestampDifference(GetCurrentTimestamp(), + finish_time, &secs, &usecs); + sleeptime = secs * 1000 + usecs / 1000; + if (WalSndDelay < sleeptime) + sleeptime = WalSndDelay; + } + else + sleeptime = WalSndDelay; + /* Sleep */ WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock, ! true, (WalSndOutTail > 0), ! sleeptime * 1000L); ! ! /* Check for replication timeout */ ! if (replication_timeout > 0 && ! standby_wal_receiver_status_interval > 0 && ! GetCurrentTimestamp() >= finish_time) ! { ! /* ! * Since typically expiration of replication timeout means ! * communication problem, we don't send the error message ! * to the standby. ! */ ! ereport(COMMERROR, ! (errmsg("terminating walsender process due to replication timeout"))); ! break; ! } } } else { /* Attempt to send the log once every loop */ ! if (!XLogSend(&caughtup, &pending)) break; } *************** *** 996,1019 **** XLogRead(char *buf, XLogRecPtr recptr, Size nbytes) * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk, * but not yet sent to the client, and send it. * - * msgbuf is a work area in which the output message is constructed. It's - * passed in just so we can avoid re-palloc'ing the buffer on each cycle. - * It must be of size 1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE. - * * If there is no unsent WAL remaining, *caughtup is set to true, otherwise * *caughtup is set to false. * * Returns true if OK, false if trouble. */ static bool ! XLogSend(char *msgbuf, bool *caughtup) { XLogRecPtr SendRqstPtr; XLogRecPtr startptr; ! XLogRecPtr endptr; Size nbytes; WalDataMessageHeader msghdr; /* * Attempt to send all data that's already been written out and fsync'd to * disk. We cannot go further than what's been written out given the --- 1163,1210 ---- * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk, * but not yet sent to the client, and send it. * * If there is no unsent WAL remaining, *caughtup is set to true, otherwise * *caughtup is set to false. * + * If there is pending WAL in output buffer, *pending is set to true, + * otherwise *pending is set to false. + * * Returns true if OK, false if trouble. */ static bool ! XLogSend(bool *caughtup, bool *pending) { XLogRecPtr SendRqstPtr; XLogRecPtr startptr; ! static XLogRecPtr endptr; Size nbytes; + uint32 n32; + int res; WalDataMessageHeader msghdr; + /* Attempt to flush pending WAL in output buffer */ + if (*pending) + { + if (WalSndOutHead != WalSndOutTail) + { + res = pq_putbytes_if_writable(WalSndOutBuffer + WalSndOutHead, + WalSndOutTail - WalSndOutHead); + if (res == EOF) + return false; + WalSndOutHead += res; + if (WalSndOutHead != WalSndOutTail) + return true; + } + + res = pq_flush_if_writable(); + if (res == EOF) + return false; + if (res == 0) + return true; + + goto updt; + } + /* * Attempt to send all data that's already been written out and fsync'd to * disk. We cannot go further than what's been written out given the *************** *** 1082,1094 **** XLogSend(char *msgbuf, bool *caughtup) /* * OK to read and send the slice. */ ! msgbuf[0] = 'w'; /* * Read the log directly into the output buffer to avoid extra memcpy * calls. */ ! XLogRead(msgbuf + 1 + sizeof(WalDataMessageHeader), startptr, nbytes); /* * We fill the message header last so that the send timestamp is taken as --- 1273,1291 ---- /* * OK to read and send the slice. */ ! WalSndOutBuffer[0] = 'd'; ! WalSndOutBuffer[5] = 'w'; ! WalSndOutHead = 0; ! WalSndOutTail = 6 + sizeof(WalDataMessageHeader) + nbytes; ! ! n32 = htonl((uint32) WalSndOutTail - 1); ! memcpy(WalSndOutBuffer + 1, &n32, 4); /* * Read the log directly into the output buffer to avoid extra memcpy * calls. */ ! XLogRead(WalSndOutBuffer + 6 + sizeof(WalDataMessageHeader), startptr, nbytes); /* * We fill the message header last so that the send timestamp is taken as *************** *** 1098,1110 **** XLogSend(char *msgbuf, bool *caughtup) msghdr.walEnd = SendRqstPtr; msghdr.sendTime = GetCurrentTimestamp(); ! memcpy(msgbuf + 1, &msghdr, sizeof(WalDataMessageHeader)); ! pq_putmessage('d', msgbuf, 1 + sizeof(WalDataMessageHeader) + nbytes); /* Flush pending output to the client */ ! if (pq_flush()) return false; sentPtr = endptr; --- 1295,1328 ---- msghdr.walEnd = SendRqstPtr; msghdr.sendTime = GetCurrentTimestamp(); ! memcpy(WalSndOutBuffer + 6, &msghdr, sizeof(WalDataMessageHeader)); ! ! res = pq_putbytes_if_writable(WalSndOutBuffer, WalSndOutTail); ! if (res == EOF) ! return false; ! WalSndOutHead = res; ! if (WalSndOutHead != WalSndOutTail) ! { ! *caughtup = false; ! *pending = true; ! return true; ! } /* Flush pending output to the client */ ! res = pq_flush_if_writable(); ! if (res == EOF) return false; + if (res == 0) + { + *caughtup = false; + *pending = true; + return true; + } + + updt: + WalSndOutHead = WalSndOutTail = 0; + *pending = false; sentPtr = endptr; *** a/src/backend/utils/misc/guc.c --- b/src/backend/utils/misc/guc.c *************** *** 1856,1861 **** static struct config_int ConfigureNamesInt[] = --- 1856,1871 ---- }, { + {"replication_timeout", PGC_SIGHUP, WAL_REPLICATION, + gettext_noop("Sets the maximum time to wait for WAL replication."), + NULL, + GUC_UNIT_MS + }, + &replication_timeout, + 0, 0, INT_MAX, NULL, NULL + }, + + { {"commit_delay", PGC_USERSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " "flushing WAL to disk."), *** a/src/backend/utils/misc/postgresql.conf.sample --- b/src/backend/utils/misc/postgresql.conf.sample *************** *** 200,205 **** --- 200,206 ---- #wal_sender_delay = 1s # walsender cycle time, 1-10000 milliseconds #wal_keep_segments = 0 # in logfile segments, 16MB each; 0 disables #vacuum_defer_cleanup_age = 0 # number of xacts by which cleanup is delayed + #replication_timeout = 0 # in milliseconds, 0 is disabled # - Standby Servers - *** a/src/include/libpq/libpq.h --- b/src/include/libpq/libpq.h *************** *** 59,65 **** extern int pq_getbyte(void); --- 59,67 ---- extern int pq_peekbyte(void); extern int pq_getbyte_if_available(unsigned char *c); extern int pq_putbytes(const char *s, size_t len); + extern int pq_putbytes_if_writable(const char *s, size_t len); extern int pq_flush(void); + extern int pq_flush_if_writable(void); extern int pq_putmessage(char msgtype, const char *s, size_t len); extern void pq_startcopyout(void); extern void pq_endcopyout(bool errorAbort); *** a/src/include/replication/walprotocol.h --- b/src/include/replication/walprotocol.h *************** *** 81,86 **** typedef struct --- 81,104 ---- } StandbyHSFeedbackMessage; /* + * GUC parameter change report from standby (message type 'g'). This is wrapped + * within a CopyData message at the FE/BE protocol level. + * + * Note that the data length is not specified here. + */ + typedef struct + { + /* + * Only change of important parameters for streaming replication needs + * to be reported. + */ + int wal_receiver_status_interval; + + /* Sender's system clock at the time of transmission */ + TimestampTz sendTime; + } StandbyGUCChangeMessage; + + /* * Maximum data payload in a WAL data message. Must be >= XLOG_BLCKSZ. * * We don't have a good idea of what a good value would be; there's some *** a/src/include/replication/walsender.h --- b/src/include/replication/walsender.h *************** *** 92,97 **** extern volatile sig_atomic_t walsender_ready_to_stop; --- 92,98 ---- /* user-settable parameters */ extern int WalSndDelay; extern int max_wal_senders; + extern int replication_timeout; extern int WalSenderMain(void); extern void WalSndSignals(void); *** a/src/include/storage/latch.h --- b/src/include/storage/latch.h *************** *** 40,46 **** extern void OwnLatch(volatile Latch *latch); extern void DisownLatch(volatile Latch *latch); extern bool WaitLatch(volatile Latch *latch, long timeout); extern int WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, ! long timeout); extern void SetLatch(volatile Latch *latch); extern void ResetLatch(volatile Latch *latch); #define TestLatch(latch) (((volatile Latch *) latch)->is_set) --- 40,46 ---- extern void DisownLatch(volatile Latch *latch); extern bool WaitLatch(volatile Latch *latch, long timeout); extern int WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, ! bool forRead, bool forWrite, long timeout); extern void SetLatch(volatile Latch *latch); extern void ResetLatch(volatile Latch *latch); #define TestLatch(latch) (((volatile Latch *) latch)->is_set)