Introduce replication timeout for asynchronous and synchronous replication.

diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 1b7f12c..5513e53 100644 (file)
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -572,16 +572,43 @@ WalSndLoop(void)
                        if ((caughtup || pending) && !got_SIGHUP && !ready_to_stop &&
                                !shutdown_requested)
                        {
+                               bool            check_timeout;
+                               long            sleeptime;
+                               int                     res;
+
                                /*
                                 * XXX: We don't really need the periodic wakeups anymore,
                                 * WaitLatchOrSocket should reliably wake up as soon as
                                 * something interesting happens.
                                 */
 
+                               /*
+                                * Check for replication timeout if it's enabled and we need to
+                                * wait for the socket to be writable to flush pending WAL in
+                                * output buffer.
+                                */
+                               check_timeout = replication_timeout > 0 && pending;
+                               if (check_timeout)
+                                       sleeptime = replication_timeout;
+                               else
+                                       sleeptime = WalSndDelay;
+
                                /* Sleep */
-                               WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
-                                                                 true, (WalSndOutTail > 0),
-                                                                 WalSndDelay * 1000L);
+                               res = WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
+                                                                               true, (WalSndOutTail > 0),
+                                                                               sleeptime * 1000L);
+
+                               if (res == 0 && check_timeout)
+                               {
+                                       /*
+                                        * 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;
+                               }
                        }
 
                        /* Process messages received from the standby */