Problem: peer can close connection before SO_NOSIGPIPE is set

Solution: setsockopt returns EINVAL if the connection was closed by
the peer after the accept returned a valid socket. This is a valid
network error and should not cause an assert.
To handle this we have to extract the setsockopt from the stream
engine, as there's no clean way to return an error from the
constructor. Instead, try to set this option before creating the
engine in the callers, and return immediately as if the accept
had failed to avoid churn. Do the same for the connect calls by
setting the option in open_socket, so that the option for that
case is set even before connecting, so there's no possible race
condition.
Since this has to be done in 4 places (tcp/ipc listener, socks
connecter and open_socket) add an utility function in ip.cpp.
Fixes #1442

src/ip.cpp

@@ -30,6 +30,7 @@
 #include "precompiled.hpp"
 #include "ip.hpp"
 #include "err.hpp"
+#include "macros.hpp"
 
 #if !defined ZMQ_HAVE_WINDOWS
 #include <fcntl.h>
@@ -46,6 +47,8 @@
 
 zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
 {
+    int rc;
+
     //  Setting this option result in sane behaviour when exec() functions
     //  are used. Old sockets are closed and don't block TCP ports etc.
 #if defined ZMQ_HAVE_SOCK_CLOEXEC
@@ -65,7 +68,7 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
     //  race condition can cause socket not to be closed (if fork happens
     //  between socket creation and this point).
 #if !defined ZMQ_HAVE_SOCK_CLOEXEC && defined FD_CLOEXEC
-    int rc = fcntl (s, F_SETFD, FD_CLOEXEC);
+    rc = fcntl (s, F_SETFD, FD_CLOEXEC);
     errno_assert (rc != -1);
 #endif
 
@@ -75,6 +78,10 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
     win_assert (brc);
 #endif
 
+    //  Socket is not yet connected so EINVAL is not a valid networking error
+    rc = zmq::set_nosigpipe (s);
+    errno_assert (rc == 0);
+
     return s;
 }
 
@@ -190,3 +197,23 @@ void zmq::set_ip_type_of_service (fd_t s_, int iptos)
     }
 #endif
 }
+
+int zmq::set_nosigpipe (fd_t s_)
+{
+#ifdef SO_NOSIGPIPE
+    //  Make sure that SIGPIPE signal is not generated when writing to a
+    //  connection that was already closed by the peer.
+    //  As per POSIX spec, EINVAL will be returned if the socket was valid but
+    //  the connection has been reset by the peer. Return an error so that the
+    //  socket can be closed and the connection retried if necessary.
+    int set = 1;
+    int rc = setsockopt (s_, SOL_SOCKET, SO_NOSIGPIPE, &set, sizeof (int));
+    if (rc != 0 && errno == EINVAL)
+        return -1;
+    errno_assert (rc == 0);
+#else
+    LIBZMQ_UNUSED (s_);
+#endif
+
+    return 0;
+}

src/ip.hpp

@@ -52,6 +52,10 @@ namespace zmq
     // Sets the IP Type-Of-Service for the underlying socket
     void set_ip_type_of_service (fd_t s_, int iptos);
 
+    // Sets the SO_NOSIGPIPE option for the underlying socket.
+    // Return 0 on success, -1 if the connection has been closed by the peer
+    int set_nosigpipe (fd_t s_);
+
 }
 
 #endif

src/ipc_listener.cpp

@@ -418,6 +418,17 @@ zmq::fd_t zmq::ipc_listener_t::accept ()
     }
 #endif
 
+    if (zmq::set_nosigpipe (sock)) {
+#ifdef ZMQ_HAVE_WINDOWS
+        int rc = closesocket (sock);
+        wsa_assert (rc != SOCKET_ERROR);
+#else
+        int rc = ::close (sock);
+        errno_assert (rc == 0);
+#endif
+        return retired_fd;
+    }
+
     return sock;
 }
 

src/stream_engine.cpp

@@ -132,13 +132,6 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_,
     }
 #endif
 
-#ifdef SO_NOSIGPIPE
-    //  Make sure that SIGPIPE signal is not generated when writing to a
-    //  connection that was already closed by the peer.
-    int set = 1;
-    rc = setsockopt (s, SOL_SOCKET, SO_NOSIGPIPE, &set, sizeof (int));
-    errno_assert (rc == 0);
-#endif
     if(options.heartbeat_interval > 0) {
         heartbeat_timeout = options.heartbeat_timeout;
         if(heartbeat_timeout == -1)

src/tcp_listener.cpp

@@ -330,6 +330,17 @@ zmq::fd_t zmq::tcp_listener_t::accept ()
         }
     }
 
+    if (zmq::set_nosigpipe (sock)) {
+#ifdef ZMQ_HAVE_WINDOWS
+        int rc = closesocket (sock);
+        wsa_assert (rc != SOCKET_ERROR);
+#else
+        int rc = ::close (sock);
+        errno_assert (rc == 0);
+#endif
+        return retired_fd;
+    }
+
     // Set the IP Type-Of-Service priority for this client socket
     if (options.tos != 0)
         set_ip_type_of_service (sock, options.tos);