Use latch instead of select() in walreceiver

Replace use of poll()/select() by WaitLatchOrSocket(), which is more
portable and flexible.

Also change walreceiver to use its procLatch instead of a custom latch.

From: Petr Jelinek <petr@2ndquadrant.com>

diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index a3921977c572d0d96aea464eed2188ccdc18426e..c7584cb1d3492e89684bf986c163e926a86d3bae 100644 (file)
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -3338,6 +3338,9 @@ pgstat_get_wait_client(WaitEventClient w)
        case WAIT_EVENT_WAL_RECEIVER_WAIT_START:
            event_name = "WalReceiverWaitStart";
            break;
+       case WAIT_EVENT_LIBPQWALRECEIVER_READ:
+           event_name = "LibPQWalReceiverRead";
+           break;
        case WAIT_EVENT_WAL_SENDER_WAIT_WAL:
            event_name = "WalSenderWaitForWAL";
            break;

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index f1c843e868c08bedf5a520df3052564f9c7fae15..6c01e7b991853a3388f11a36db605cb2abbb61eb 100644 (file)
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -23,19 +23,11 @@
 #include "pqexpbuffer.h"
 #include "access/xlog.h"
 #include "miscadmin.h"
+#include "pgstat.h"
 #include "replication/walreceiver.h"
+#include "storage/proc.h"
 #include "utils/builtins.h"
 
-#ifdef HAVE_POLL_H
-#include <poll.h>
-#endif
-#ifdef HAVE_SYS_POLL_H
-#include <sys/poll.h>
-#endif
-#ifdef HAVE_SYS_SELECT_H
-#include <sys/select.h>
-#endif
-
 PG_MODULE_MAGIC;
 
 void       _PG_init(void);
@@ -59,7 +51,6 @@ static void libpqrcv_send(const char *buffer, int nbytes);
 static void libpqrcv_disconnect(void);
 
 /* Prototypes for private functions */
-static bool libpq_select(int timeout_ms);
 static PGresult *libpqrcv_PQexec(const char *query);
 
 /*
@@ -366,67 +357,6 @@ libpqrcv_readtimelinehistoryfile(TimeLineID tli,
    PQclear(res);
 }
 
-/*
- * Wait until we can read WAL stream, or timeout.
- *
- * Returns true if data has become available for reading, false if timed out
- * or interrupted by signal.
- *
- * This is based on pqSocketCheck.
- */
-static bool
-libpq_select(int timeout_ms)
-{
-   int         ret;
-
-   Assert(streamConn != NULL);
-   if (PQsocket(streamConn) < 0)
-       ereport(ERROR,
-               (errcode_for_socket_access(),
-                errmsg("invalid socket: %s", PQerrorMessage(streamConn))));
-
-   /* We use poll(2) if available, otherwise select(2) */
-   {
-#ifdef HAVE_POLL
-       struct pollfd input_fd;
-
-       input_fd.fd = PQsocket(streamConn);
-       input_fd.events = POLLIN | POLLERR;
-       input_fd.revents = 0;
-
-       ret = poll(&input_fd, 1, timeout_ms);
-#else                          /* !HAVE_POLL */
-
-       fd_set      input_mask;
-       struct timeval timeout;
-       struct timeval *ptr_timeout;
-
-       FD_ZERO(&input_mask);
-       FD_SET(PQsocket(streamConn), &input_mask);
-
-       if (timeout_ms < 0)
-           ptr_timeout = NULL;
-       else
-       {
-           timeout.tv_sec = timeout_ms / 1000;
-           timeout.tv_usec = (timeout_ms % 1000) * 1000;
-           ptr_timeout = &timeout;
-       }
-
-       ret = select(PQsocket(streamConn) + 1, &input_mask,
-                    NULL, NULL, ptr_timeout);
-#endif   /* HAVE_POLL */
-   }
-
-   if (ret == 0 || (ret < 0 && errno == EINTR))
-       return false;
-   if (ret < 0)
-       ereport(ERROR,
-               (errcode_for_socket_access(),
-                errmsg("select() failed: %m")));
-   return true;
-}
-
 /*
  * Send a query and wait for the results by using the asynchronous libpq
  * functions and the backend version of select().
@@ -470,14 +400,31 @@ libpqrcv_PQexec(const char *query)
         */
        while (PQisBusy(streamConn))
        {
+           int         rc;
+
            /*
             * We don't need to break down the sleep into smaller increments,
-            * and check for interrupts after each nap, since we can just
-            * elog(FATAL) within SIGTERM signal handler if the signal arrives
-            * in the middle of establishment of replication connection.
+            * since we'll get interrupted by signals and can either handle
+            * interrupts here or elog(FATAL) within SIGTERM signal handler if
+            * the signal arrives in the middle of establishment of
+            * replication connection.
             */
-           if (!libpq_select(-1))
-               continue;       /* interrupted */
+           ResetLatch(&MyProc->procLatch);
+           rc = WaitLatchOrSocket(&MyProc->procLatch,
+                                  WL_POSTMASTER_DEATH | WL_SOCKET_READABLE |
+                                  WL_LATCH_SET,
+                                  PQsocket(streamConn),
+                                  0,
+                                  WAIT_EVENT_LIBPQWALRECEIVER_READ);
+           if (rc & WL_POSTMASTER_DEATH)
+               exit(1);
+
+           /* interrupted */
+           if (rc & WL_LATCH_SET)
+           {
+               CHECK_FOR_INTERRUPTS();
+               continue;
+           }
            if (PQconsumeInput(streamConn) == 0)
                return NULL;    /* trouble */
        }

diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 2bb3dce1b1c5bab7ac7d1e1aeb59e6e50b7bc7a3..8bfb041560823f2c3ca11ba09b42b7ece190c09f 100644 (file)
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -261,7 +261,7 @@ WalReceiverMain(void)
    /* Arrange to clean up at walreceiver exit */
    on_shmem_exit(WalRcvDie, 0);
 
-   OwnLatch(&walrcv->latch);
+   walrcv->latch = &MyProc->procLatch;
 
    /* Properly accept or ignore signals the postmaster might send us */
    pqsignal(SIGHUP, WalRcvSigHupHandler);      /* set flag to read config
@@ -483,7 +483,7 @@ WalReceiverMain(void)
                 * avoiding some system calls.
                 */
                Assert(wait_fd != PGINVALID_SOCKET);
-               rc = WaitLatchOrSocket(&walrcv->latch,
+               rc = WaitLatchOrSocket(walrcv->latch,
                                   WL_POSTMASTER_DEATH | WL_SOCKET_READABLE |
                                       WL_TIMEOUT | WL_LATCH_SET,
                                       wait_fd,
@@ -491,7 +491,7 @@ WalReceiverMain(void)
                                       WAIT_EVENT_WAL_RECEIVER_MAIN);
                if (rc & WL_LATCH_SET)
                {
-                   ResetLatch(&walrcv->latch);
+                   ResetLatch(walrcv->latch);
                    if (walrcv->force_reply)
                    {
                        /*
@@ -652,7 +652,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
    WakeupRecovery();
    for (;;)
    {
-       ResetLatch(&walrcv->latch);
+       ResetLatch(walrcv->latch);
 
        /*
         * Emergency bailout if postmaster has died.  This is to avoid the
@@ -687,7 +687,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
        }
        SpinLockRelease(&walrcv->mutex);
 
-       WaitLatch(&walrcv->latch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
+       WaitLatch(walrcv->latch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
                  WAIT_EVENT_WAL_RECEIVER_WAIT_START);
    }
 
@@ -763,7 +763,7 @@ WalRcvDie(int code, Datum arg)
    /* Ensure that all WAL records received are flushed to disk */
    XLogWalRcvFlush(true);
 
-   DisownLatch(&walrcv->latch);
+   walrcv->latch = NULL;
 
    SpinLockAcquire(&walrcv->mutex);
    Assert(walrcv->walRcvState == WALRCV_STREAMING ||
@@ -812,7 +812,8 @@ WalRcvShutdownHandler(SIGNAL_ARGS)
 
    got_SIGTERM = true;
 
-   SetLatch(&WalRcv->latch);
+   if (WalRcv->latch)
+       SetLatch(WalRcv->latch);
 
    /* Don't joggle the elbow of proc_exit */
    if (!proc_exit_inprogress && WalRcvImmediateInterruptOK)
@@ -1297,7 +1298,8 @@ void
 WalRcvForceReply(void)
 {
    WalRcv->force_reply = true;
-   SetLatch(&WalRcv->latch);
+   if (WalRcv->latch)
+       SetLatch(WalRcv->latch);
 }
 
 /*

diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 5f6e423f1f63c74d53ed7607849a9bf81ac6eae9..01111a4c12bcd8f3ddde0ebe47edeab4ab3c304c 100644 (file)
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -64,7 +64,7 @@ WalRcvShmemInit(void)
        MemSet(WalRcv, 0, WalRcvShmemSize());
        WalRcv->walRcvState = WALRCV_STOPPED;
        SpinLockInit(&WalRcv->mutex);
-       InitSharedLatch(&WalRcv->latch);
+       WalRcv->latch = NULL;
    }
 }
 
@@ -279,8 +279,8 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
 
    if (launch)
        SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
-   else
-       SetLatch(&walrcv->latch);
+   else if (walrcv->latch)
+       SetLatch(walrcv->latch);
 }
 
 /*

diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 0b85b7ad3ae9975eb20e6803c86c498e037ff0de..152ff0620852133323edda179e17539893fbe539 100644 (file)
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -763,6 +763,7 @@ typedef enum
    WAIT_EVENT_CLIENT_WRITE,
    WAIT_EVENT_SSL_OPEN_SERVER,
    WAIT_EVENT_WAL_RECEIVER_WAIT_START,
+   WAIT_EVENT_LIBPQWALRECEIVER_READ,
    WAIT_EVENT_WAL_SENDER_WAIT_WAL,
    WAIT_EVENT_WAL_SENDER_WRITE_DATA
 } WaitEventClient;

diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index cd787c92b3fc9eb2fb8de62d1199ceb64cc6c1c1..afbb8d8b9541e450d08702834406cc7b835be80c 100644 (file)
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -127,8 +127,9 @@ typedef struct
     * where to start streaming (after setting receiveStart and
     * receiveStartTLI), and also to tell it to send apply feedback to the
     * primary whenever specially marked commit records are applied.
+    * This is normally mapped to procLatch when walreceiver is running.
     */
-   Latch       latch;
+   Latch      *latch;
 } WalRcvData;
 
 extern WalRcvData *WalRcv;