[PATCH 2/2 v3] [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>

---
src/interfaces/libpq/fe-connect.c | 40 ++++++++++++++++++
src/interfaces/libpq/fe-secure.c | 83 +++++++++++++++++++++++++++++---------
src/interfaces/libpq/libpq-int.h | 2
3 files changed, 107 insertions(+), 18 deletions(-)

*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
***************
*** 1085,1090 **** keep_going: /* We will come back to here until there is
--- 1085,1091 ----
while (conn->addr_cur != NULL)
{
struct addrinfo *addr_cur = conn->addr_cur;
+ int optval;

/* Remember current address for possible error msg */
memcpy(&conn->raddr.addr, addr_cur->ai_addr,
***************
*** 1149,1154 **** keep_going: /* We will come back to here until there is
--- 1150,1194 ----
}
#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
***************
*** 122,127 **** static long win32_ssl_create_mutex = 0;
--- 122,139 ----
*/

#ifndef WIN32
+
+ static inline int sigpipe_masked(PGconn *conn)
+ {
+ /* If we're on an SSL connection, we can only use SO_NOSIGPIPE masking.
+ * Otherwise, we can handle SO_NOSIGPIPE or the MSG_NOSIGNAL flag */
+ #ifdef USE_SSL
+ if (conn->ssl)
+ return conn->sigpipe_so;
+ #endif
+ return conn->sigpipe_so || conn->sigpipe_flag;
+ }
+
#ifdef ENABLE_THREAD_SAFETY

struct sigpipe_info {
***************
*** 130,137 **** struct sigpipe_info {
bool got_epipe;
};

! static inline int disable_sigpipe(struct sigpipe_info *info)
{
info->got_epipe = false;
return pq_block_sigpipe(&info->oldsigmask, &info->sigpipe_pending) < 0;
}
--- 142,152 ----
bool got_epipe;
};

! static inline int disable_sigpipe(PGconn *conn, struct sigpipe_info *info)
{
+ if (sigpipe_masked(conn))
+ return 0;
+
info->got_epipe = false;
return pq_block_sigpipe(&info->oldsigmask, &info->sigpipe_pending) < 0;
}
***************
*** 142,149 **** static inline void remember_epipe(struct sigpipe_info *info, bool cond)
info->got_epipe = true;
}

! static inline void restore_sigpipe(struct sigpipe_info *info)
{
pq_reset_sigpipe(&info->oldsigmask, info->sigpipe_pending, info->got_epipe);
}

--- 157,167 ----
info->got_epipe = true;
}

! static inline void restore_sigpipe(PGconn *conn, struct sigpipe_info *info)
{
+ if (sigpipe_masked(conn))
+ return;
+
pq_reset_sigpipe(&info->oldsigmask, info->sigpipe_pending, info->got_epipe);
}

***************
*** 153,161 **** struct sigpipe_info {
pqsigfunc oldhandler;
};

! static inline int disable_sigpipe(struct sigpipe_info *info)
{
! info->oldhandler = pqsignal(SIGPIPE, SIG_IGN);
return 0;
}

--- 171,180 ----
pqsigfunc oldhandler;
};

! static inline int disable_sigpipe(PGconn *conn, struct sigpipe_info *info)
{
! if (!sigpipe_masked(conn))
! info->oldhandler = pqsignal(SIGPIPE, SIG_IGN);
return 0;
}

***************
*** 163,180 **** static inline void remember_epipe(struct sigpipe_info *info, bool cond)
{
}

! static inline void restore_sigpipe(struct sigpipe_info *info)
{
! pqsignal(SIGPIPE, info->oldhandler);
}

#endif /* ENABLE_THREAD_SAFETY */
#else /* WIN32 */

struct sigpipe_info { };
! static inline int disable_sigpipe(struct sigpipe_info *info) { return 0; }
static inline void remember_epipe(struct sigpipe_info *info, bool cond) { }
! static inline void restore_sigpipe(struct sigpipe_info *info) { }

#endif /* WIN32 */

--- 182,203 ----
{
}

! static inline void restore_sigpipe(PGconn *conn, struct sigpipe_info *info)
{
! if (!sigpipe_masked(conn))
! pqsignal(SIGPIPE, info->oldhandler);
}

#endif /* ENABLE_THREAD_SAFETY */
#else /* WIN32 */

struct sigpipe_info { };
! static inline int disable_sigpipe(PGConn *conn, struct sigpipe_info *info)
! {
! return 0;
! }
static inline void remember_epipe(struct sigpipe_info *info, bool cond) { }
! static inline void restore_sigpipe(PGConn *conn, struct sigpipe_info *info) { }

#endif /* WIN32 */

***************
*** 306,312 **** pqsecure_read(PGconn *conn, void *ptr, size_t len)
struct sigpipe_info info;

/* SSL_read can write to the socket, so we need to disable SIGPIPE */
! if (disable_sigpipe(&info))
return -1;

rloop:
--- 329,335 ----
struct sigpipe_info info;

/* SSL_read can write to the socket, so we need to disable SIGPIPE */
! if (disable_sigpipe(conn, &info))
return -1;

rloop:
***************
*** 370,376 **** rloop:
break;
}

! restore_sigpipe(&info);
}
else
#endif
--- 393,399 ----
break;
}

! restore_sigpipe(conn, &info);
}
else
#endif
***************
*** 388,401 **** pqsecure_write(PGconn *conn, const void *ptr, size_t len)
ssize_t n;
struct sigpipe_info info;

- if (disable_sigpipe(&info))
- 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)
--- 411,424 ----
ssize_t n;
struct 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)
***************
*** 458,468 **** pqsecure_write(PGconn *conn, const void *ptr, size_t len)
else
#endif
{
! n = send(conn->sock, ptr, len, 0);
! remember_epipe(&info, n < 0 && SOCK_ERRNO == EPIPE);
}

! restore_sigpipe(&info);

return n;
}
--- 481,515 ----
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;
}
***************
*** 1219,1232 **** close_SSL(PGconn *conn)
if (conn->ssl)
{
struct sigpipe_info info;
! disable_sigpipe(&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(&info);
}

if (conn->peer)
--- 1266,1279 ----
if (conn->ssl)
{
struct 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
***************
*** 336,341 **** struct pg_conn
--- 336,343 ----
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 */