[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(at)ozlabs(dot)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 of SIGPIPE 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 common to 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 common to 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 establishing connection */
struct addrinfo *addrlist; /* list of possible backend addresses */