[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 

---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 */