[PATCH v4] [libpq] Try to avoid manually masking SIGPIPEs on every send()

Currently, libpq will wrap each send() call on the connection with
two system calls to mask SIGPIPEs. This results in 3 syscalls instead
of one, and (on Linux) can lead to high contention on the signal
mask locks in threaded apps.

We have a couple of other methods to avoid SIGPIPEs:
sockopt(SO_NOSIGPIPE) and the MSG_NOSIGNAL flag to send().

This change attempts to use these if they're available at compile-
and run-time. If not, we drop back to manipulating the signal mask as
before.

Signed-off-by: Jeremy Kerr <jk@ozlabs.org>

---
v4: roll into one patch, use macros

---src/interfaces/libpq/fe-connect.c |   42 ++++++++++++src/interfaces/libpq/fe-secure.c  |  131
++++++++++++++++++++++++++------------src/interfaces/libpq/libpq-int.h |    2 3 files changed, 136 insertions(+), 39
deletions(-)

*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
***************
*** 1089,1094 **** keep_going:                        /* We will come back to here until there is
--- 1089,1097 ----                 while (conn->addr_cur != NULL)                 {                     struct addrinfo
*addr_cur= conn->addr_cur;
 
+ #ifdef SO_NOSIGPIPE
+                     int optval;
+ #endif /* SO_NOSIGPIPE */                      /* Remember current address for possible error msg */
  memcpy(&conn->raddr.addr, addr_cur->ai_addr,
 
***************
*** 1153,1158 **** keep_going:                        /* We will come back to here until there is
--- 1156,1200 ----                     } #endif   /* F_SETFD */ 
+                     /* We have three methods of blocking sigpipe during
+                      * send() calls to this socket:
+                      *
+                      *  - setsockopt(sock, SO_NOSIGPIPE)
+                      *  - send(sock, ..., MSG_NOSIGNAL)
+                      *  - setting the signal mask to SIG_IGN during send()
+                      *
+                      * The first two reduce the number of syscalls (for the
+                      * third, we require three syscalls to implement a send()),
+                      * so use them if they're available. Their availability is
+                      * flagged in the following members of PGconn:
+                      *
+                      * conn->sigpipe_so        - we have set up SO_NOSIGPIPE
+                      * conn->sigpipe_flag    - we're specifying MSG_NOSIGNAL
+                      *
+                      * If we can use SO_NOSIGPIPE, then set sigpipe_so here and
+                      * we don't need to care about anything else. Otherwise,
+                      * try MSG_NOSIGNAL by setting sigpipe_flag. If we get an
+                      * error with MSG_NOSIGNAL, we clear the flag and revert
+                      * to manual masking.
+                      */
+                     conn->sigpipe_so = false;
+ #ifdef MSG_NOSIGNAL
+                     conn->sigpipe_flag = true;
+ #else /* !MSG_NOSIGNAL */
+                     conn->sigpipe_flag = false;
+ #endif /* MSG_NOSIGNAL */
+ 
+ #ifdef SO_NOSIGPIPE
+                     optval = 1;
+                     if (!setsockopt(conn->sock, SOL_SOCKET, SO_NOSIGPIPE,
+                             (char *)&optval, sizeof(optval)))
+                     {
+                         conn->sigpipe_so = true;
+                         conn->sigpipe_flag = false;
+                     }
+ #endif /* SO_NOSIGPIPE */
+ 
+                      /*                      * Start/make connection.  This should not block, since we
     * are in nonblock mode.  If it does, well, too bad.
 
*** a/src/interfaces/libpq/fe-secure.c
--- b/src/interfaces/libpq/fe-secure.c
***************
*** 118,161 **** static long win32_ssl_create_mutex = 0;  /*  * Macros to handle disabling and then restoring the state
ofSIGPIPE handling.
 
-  * Note that DISABLE_SIGPIPE() must appear at the start of a block.  */  #ifndef WIN32 #ifdef ENABLE_THREAD_SAFETY 
! #define DISABLE_SIGPIPE(failaction) \
!     sigset_t    osigmask; \
!     bool        sigpipe_pending; \
!     bool        got_epipe = false; \
! \
!     if (pq_block_sigpipe(&osigmask, &sigpipe_pending) < 0) \
!         failaction 
! #define REMEMBER_EPIPE(cond) \
!     do { \
!         if (cond) \
!             got_epipe = true; \
!     } while (0) 
! #define RESTORE_SIGPIPE() \
!     pq_reset_sigpipe(&osigmask, sigpipe_pending, got_epipe)
! #else                            /* !ENABLE_THREAD_SAFETY */ 
! #define DISABLE_SIGPIPE(failaction) \
!     pqsigfunc    oldsighandler = pqsignal(SIGPIPE, SIG_IGN) 
! #define REMEMBER_EPIPE(cond) 
! #define RESTORE_SIGPIPE() \
!     pqsignal(SIGPIPE, oldsighandler)
! #endif   /* ENABLE_THREAD_SAFETY */
! #else                            /* WIN32 */ 
! #define DISABLE_SIGPIPE(failaction)
! #define REMEMBER_EPIPE(cond)
! #define RESTORE_SIGPIPE()
! #endif   /* WIN32 */  /* ------------------------------------------------------------ */ /*             Procedures
commonto all secure sessions            */
 
--- 118,185 ----  /*  * Macros to handle disabling and then restoring the state of SIGPIPE handling.  */  #ifndef
WIN32
+ 
+ #ifdef USE_SSL
+ #define SIGPIPE_MASKED(c) \
+     ((c)->ssl ? (c)->sigpipe_so : (c)->sigpipe_so || (c)->sigpipe_flag)
+ #else
+ #define SIGPIPE_MASKED(c) \
+     ((c)->sigpipe_so || (c)->sigpipe_flag)
+ #endif
+  #ifdef ENABLE_THREAD_SAFETY 
! struct sigpipe_info {
!     sigset_t    oldsigmask;
!     bool        sigpipe_pending;
!     bool        got_epipe;
! }; 
! #define DECLARE_SIGPIPE_INFO(var) struct sigpipe_info var 
! #define DISABLE_SIGPIPE(conn, info) \
!     ({ int __x = 0; \
!         if (!SIGPIPE_MASKED(conn)) \
!             (info).got_epipe = false; \
!             __x = pq_block_sigpipe(&(info).oldsigmask, \
!                     &(info).sigpipe_pending) < 0; \
!         __x; }) 
! #define REMEMBER_EPIPE(info, cond) \
!     if (cond) (info).got_epipe = true 
! #define RESTORE_SIGPIPE(conn, info) \
!     if (!SIGPIPE_MASKED(conn)) \
!         pq_reset_sigpipe(&(info).oldsigmask, (info).sigpipe_pending, \
!                          (info).got_epipe) 
! #else    /* !ENABLE_THREAD_SAFETY */
! 
! #define DECLARE_SIGPIPE_INFO(var) pqsigfunc var = NULL
! 
! #define DISABLE_SIGPIPE(conn, info) \
!     ({ if (!SIGPIPE_MASKED(conn)) \
!         (info) = pqsignal(SIGPIPE, SIG_IGN); \
!       0; })
! 
! #define REMEMBER_EPIPE(info, cond)
! 
! #define RESTORE_SIGPIPE(conn, info) \
!     if (!SIGPIPE_MASKED(conn)) \
!         pqsignal(SIGPIPE, (info))
! 
! #endif    /* ENABLE_THREAD_SAFETY */
! #else    /* WIN32 */
! 
! #define DECLARE_SIGPIPE_INFO(var)
! #define DISABLE_SIGPIPE(conn, info) 0
! #define REMEMBER_EPIPE(info, cond)
! #define RESTORE_SIGPIPE(conn, info) 
! #endif    /* WIN32 */  /* ------------------------------------------------------------ */ /*             Procedures
commonto all secure sessions            */
 
***************
*** 283,291 **** pqsecure_read(PGconn *conn, void *ptr, size_t len)     if (conn->ssl)     {         int
err;         /* SSL_read can write to the socket, so we need to disable SIGPIPE */
 
!         DISABLE_SIGPIPE(return -1);  rloop:         n = SSL_read(conn->ssl, ptr, len);
--- 307,317 ----     if (conn->ssl)     {         int            err;
+         DECLARE_SIGPIPE_INFO(info);          /* SSL_read can write to the socket, so we need to disable SIGPIPE */
!         if (DISABLE_SIGPIPE(conn, info))
!             return -1;  rloop:         n = SSL_read(conn->ssl, ptr, len);
***************
*** 312,318 **** rloop:                      if (n == -1)                     {
!                         REMEMBER_EPIPE(SOCK_ERRNO == EPIPE);
printfPQExpBuffer(&conn->errorMessage,                                    libpq_gettext("SSL SYSCALL error: %s\n"),
                       SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf)));
 
--- 338,344 ----                      if (n == -1)                     {
!                         REMEMBER_EPIPE(info, SOCK_ERRNO == EPIPE);
printfPQExpBuffer(&conn->errorMessage,                                    libpq_gettext("SSL SYSCALL error: %s\n"),
                       SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf)));
 
***************
*** 348,354 **** rloop:                 break;         } 
!         RESTORE_SIGPIPE();     }     else #endif
--- 374,380 ----                 break;         } 
!         RESTORE_SIGPIPE(conn, info);     }     else #endif
***************
*** 364,377 **** ssize_t pqsecure_write(PGconn *conn, const void *ptr, size_t len) {     ssize_t        n;
! 
!     DISABLE_SIGPIPE(return -1);  #ifdef USE_SSL     if (conn->ssl)     {         int            err;          n =
SSL_write(conn->ssl,ptr, len);         err = SSL_get_error(conn->ssl, n);         switch (err)
 
--- 390,405 ---- pqsecure_write(PGconn *conn, const void *ptr, size_t len) {     ssize_t        n;
!     DECLARE_SIGPIPE_INFO(info);  #ifdef USE_SSL     if (conn->ssl)     {         int            err; 
+         if (DISABLE_SIGPIPE(conn, info))
+             return -1;
+          n = SSL_write(conn->ssl, ptr, len);         err = SSL_get_error(conn->ssl, n);         switch (err)
***************
*** 396,402 **** pqsecure_write(PGconn *conn, const void *ptr, size_t len)                      if (n == -1)
        {
 
!                         REMEMBER_EPIPE(SOCK_ERRNO == EPIPE);
printfPQExpBuffer(&conn->errorMessage,                                    libpq_gettext("SSL SYSCALL error: %s\n"),
                       SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf)));
 
--- 424,430 ----                      if (n == -1)                     {
!                         REMEMBER_EPIPE(&info, SOCK_ERRNO == EPIPE);
printfPQExpBuffer(&conn->errorMessage,                                    libpq_gettext("SSL SYSCALL error: %s\n"),
                       SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf)));
 
***************
*** 434,444 **** pqsecure_write(PGconn *conn, const void *ptr, size_t len)     else #endif     {
!         n = send(conn->sock, ptr, len, 0);
!         REMEMBER_EPIPE(n < 0 && SOCK_ERRNO == EPIPE);     } 
!     RESTORE_SIGPIPE();      return n; }
--- 462,496 ----     else #endif     {
!         int flags = 0;
! 
! #ifdef MSG_NOSIGNAL
!         if (!conn->sigpipe_so && conn->sigpipe_flag)
!             flags |= MSG_NOSIGNAL;
! #endif /* MSG_NOSIGNAL */
! 
! retry_masked:
!         if (DISABLE_SIGPIPE(conn, info))
!             return -1;
! 
!         n = send(conn->sock, ptr, len, flags);
! 
!         if (n < 0) {
!             /* if we see an EINVAL, it may be because MSG_NOSIGNAL isn't
!              * available on this machine. So, clear sigpipe_flag so we don't
!              * try this flag again, and retry the send().
!              */
!             if (flags != 0 && SOCK_ERRNO == EINVAL) {
!                 conn->sigpipe_flag = false;
!                 flags = 0;
!                 goto retry_masked;
!             }
! 
!             REMEMBER_EPIPE(info, SOCK_ERRNO == EPIPE);
!         }     } 
!     RESTORE_SIGPIPE(conn, info);      return n; }
***************
*** 1220,1233 **** close_SSL(PGconn *conn) {     if (conn->ssl)     {
!         DISABLE_SIGPIPE((void) 0);         SSL_shutdown(conn->ssl);         SSL_free(conn->ssl);         conn->ssl =
NULL;        pqsecure_destroy();         /* We have to assume we got EPIPE */
 
!         REMEMBER_EPIPE(true);
!         RESTORE_SIGPIPE();     }      if (conn->peer)
--- 1272,1286 ---- {     if (conn->ssl)     {
!         DECLARE_SIGPIPE_INFO(info);
!         DISABLE_SIGPIPE(conn, info);         SSL_shutdown(conn->ssl);         SSL_free(conn->ssl);         conn->ssl
=NULL;         pqsecure_destroy();         /* We have to assume we got EPIPE */
 
!         REMEMBER_EPIPE(info, true);
!         RESTORE_SIGPIPE(conn, info);     }      if (conn->peer)
*** a/src/interfaces/libpq/libpq-int.h
--- b/src/interfaces/libpq/libpq-int.h
***************
*** 341,346 **** struct pg_conn
--- 341,348 ----     ProtocolVersion pversion;    /* FE/BE protocol version in use */     int            sversion;
 /* server version, e.g. 70401 for 7.4.1 */     bool        password_needed;    /* true if server demanded a password
*/
+     bool        sigpipe_so;        /* have we masked sigpipes via SO_NOSIGPIPE? */
+     bool        sigpipe_flag;    /* can we mask sigpipes via MSG_NOSIGNAL? */      /* Transient state needed while
establishingconnection */     struct addrinfo *addrlist;    /* list of possible backend addresses */