Set options on new sockets in systematic manner

This patch consolidates the up-to-now scattered code that
sets different options on newly created sockets.

There are open_socket and open_tcp_socket functions that
do the tuning automatically.

In case the socket is not created but got from elsewhere
(such as accept() call) there are tune_socket and tune_tcp_socket
functions that will do the tuning.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

6 files changed, 44 insertions, 61 deletions

diff --git a/src/ip.cpp b/src/ip.cpp
index e11ec45..a54d75e 100644
--- a/src/ip.cpp
+++ b/src/ip.cpp
@@ -46,24 +46,35 @@ xs::fd_t xs::open_socket (int domain_, int type_, int protocol_)
 #endif
 
     fd_t s = socket (domain_, type_, protocol_);
-    if (s == retired_fd)
+    if (s == retired_fd) {
+#ifdef XS_HAVE_WINDOWS
+        wsa_error_to_errno ();
+#endif
         return retired_fd;
+    }
+    tune_socket (s);
+    return s;
+}
 
-    //  If there's no SOCK_CLOEXEC, let's try the second best option. Note that
-    //  race condition can cause socket not to be closed (if fork happens
-    //  between socket creation and this point).
-#if !defined XS_HAVE_SOCK_CLOEXEC && defined FD_CLOEXEC
-    int rc = fcntl (s, F_SETFD, FD_CLOEXEC);
+void xs::tune_socket (fd_t s_)
+{
+    //  Prevent socket to be inherited by child processes.
+#if defined FD_CLOEXEC
+    int rc = fcntl (s_, F_SETFD, FD_CLOEXEC);
     errno_assert (rc != -1);
 #endif
-
-    //  On Windows, preventing sockets to be inherited by child processes is
-    //  done using SetHandleInformation function.
 #if defined XS_HAVE_WINDOWS && defined HANDLE_FLAG_INHERIT
-    BOOL brc = SetHandleInformation ((HANDLE) s, HANDLE_FLAG_INHERIT, 0);
+    BOOL brc = SetHandleInformation ((HANDLE) s_, HANDLE_FLAG_INHERIT, 0);
     win_assert (brc);
 #endif
+}
 
+xs::fd_t xs::open_tcp_socket (int domain_, bool keepalive_)
+{
+    fd_t s = open_socket (domain_, SOCK_STREAM, IPPROTO_TCP);
+    if (s == retired_fd)
+        return retired_fd;
+    tune_tcp_socket (s, keepalive_);
     return s;
 }
 
diff --git a/src/ip.hpp b/src/ip.hpp
index 9487303..da1fe84 100644
--- a/src/ip.hpp
+++ b/src/ip.hpp
@@ -28,9 +28,15 @@ namespace xs
 {
 
     //  Same as socket(2), but allows for transparent tweaking the options.
+    //  These functions automatically tune the socket so there's no need to
+    //  call tune_socket/tune_tcp_socket afterwards.
     fd_t open_socket (int domain_, int type_, int protocol_);
+    fd_t open_tcp_socket (int domain_, bool keepalive_);
 
-    //  Tunes the supplied TCP socket for the best latency.
+    //  Tunes the supplied socket. Use these functions if you've got
+    //  the socket in some other way, not by open_socket/open_tcp_socket
+    //  (e.g. using accept()).
+    void tune_socket (fd_t s_);
     void tune_tcp_socket (fd_t s_, bool keeaplive_);
 
     //  Sets the socket into non-blocking mode.
diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp
index 743ca86..43f75c1 100644
--- a/src/ipc_listener.cpp
+++ b/src/ipc_listener.cpp
@@ -162,6 +162,7 @@ xs::fd_t xs::ipc_listener_t::accept ()
             errno == ENFILE);
         return retired_fd;
     }
+    tune_socket (sock);
     return sock;
 }
 
diff --git a/src/signaler.cpp b/src/signaler.cpp
index 4d983a8..1dbd9bb 100644
--- a/src/signaler.cpp
+++ b/src/signaler.cpp
@@ -288,17 +288,13 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
 
     //  Create listening socket.
     SOCKET listener;
-    listener = open_socket (AF_INET, SOCK_STREAM, 0);
+    listener = open_tcp_socket (AF_INET, false);
     wsa_assert (listener != INVALID_SOCKET);
 
-    //  Set SO_REUSEADDR and TCP_NODELAY on listening socket.
-    BOOL so_reuseaddr = 1;
+    //  Set SO_REUSEADDR on the listening socket.
+    BOOL reuseaddr = 1;
     int rc = setsockopt (listener, SOL_SOCKET, SO_REUSEADDR,
-        (char *)&so_reuseaddr, sizeof (so_reuseaddr));
-    wsa_assert (rc != SOCKET_ERROR);
-    BOOL tcp_nodelay = 1;
-    rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY,
-        (char *)&tcp_nodelay, sizeof (tcp_nodelay));
+        (char*) &reuseaddr, sizeof (reuseaddr));
     wsa_assert (rc != SOCKET_ERROR);
 
     //  Bind listening socket to the local port.
@@ -315,14 +311,9 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     wsa_assert (rc != SOCKET_ERROR);
 
     //  Create the writer socket.
-    *w_ = WSASocket (AF_INET, SOCK_STREAM, 0, NULL, 0,  0);
+    *w_ = open_tcp_socket (AF_INET, false);
     wsa_assert (*w_ != INVALID_SOCKET);
 
-    //  Set TCP_NODELAY on writer socket.
-    rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY,
-        (char *)&tcp_nodelay, sizeof (tcp_nodelay));
-    wsa_assert (rc != SOCKET_ERROR);
-
     //  Connect writer to the listener.
     rc = connect (*w_, (sockaddr *) &addr, sizeof (addr));
     wsa_assert (rc != SOCKET_ERROR);
@@ -330,6 +321,7 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     //  Accept connection from writer.
     *r_ = accept (listener, NULL, NULL);
     wsa_assert (*r_ != INVALID_SOCKET);
+    tune_tcp_socket (*r_, false);
 
     //  We don't need the listening socket anymore. Close it.
     rc = closesocket (listener);
@@ -355,17 +347,14 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     lcladdr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
     lcladdr.sin_port = 0;
 
-    int listener = open_socket (AF_INET, SOCK_STREAM, 0);
+    int listener = open_tcp_socket (AF_INET, false);
     errno_assert (listener != -1);
 
     int on = 1;
-    int rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on));
-    errno_assert (rc != -1);
-
     rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on));
     errno_assert (rc != -1);
 
-    rc = bind(listener, (struct sockaddr*) &lcladdr, sizeof (lcladdr));
+    rc = bind (listener, (struct sockaddr*) &lcladdr, sizeof (lcladdr));
     errno_assert (rc != -1);
 
     socklen_t lcladdr_len = sizeof (lcladdr);
@@ -376,12 +365,9 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     rc = listen (listener, 1);
     errno_assert (rc != -1);
 
-    *w_ = open_socket (AF_INET, SOCK_STREAM, 0);
+    *w_ = open_tcp_socket (AF_INET, false);
     errno_assert (*w_ != -1);
 
-    rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on));
-    errno_assert (rc != -1);
-
     rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on));
     errno_assert (rc != -1);
 
@@ -390,6 +376,7 @@ int xs::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
 
     *r_ = accept (listener, NULL, NULL);
     errno_assert (*r_ != -1);
+    tune_tcp_socket (*r_);
 
     close (listener);
 
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index 8072209..2fbbff6 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -107,8 +107,6 @@ void xs::tcp_connecter_t::out_event (fd_t fd_)
         return;
     }
 
-    tune_tcp_socket (fd, options.keepalive ? true : false);
-
     //  Create the engine object for this connection.
     stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options);
     alloc_assert (engine);
@@ -207,16 +205,9 @@ int xs::tcp_connecter_t::open ()
     xs_assert (s == retired_fd);
 
     //  Create the socket.
-    s = open_socket (address.family (), SOCK_STREAM, IPPROTO_TCP);
-#ifdef XS_HAVE_WINDOWS
-    if (s == INVALID_SOCKET) {
-        wsa_error_to_errno ();
-        return -1;
-    }
-#else
+    s = open_tcp_socket (address.family (), options.keepalive ? true : false);
     if (s == -1)
         return -1;
-#endif
 
     //  On some systems, IPv4 mapping in IPv6 sockets is disabled by default.
     //  Switch it on in such cases.
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index 2fda819..22797e2 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -82,12 +82,9 @@ void xs::tcp_listener_t::in_event (fd_t fd_)
     fd_t fd = accept ();
 
     //  If connection was reset by the peer in the meantime, just ignore it.
-    //  TODO: Handle specific errors like ENFILE/EMFILE etc.
     if (fd == retired_fd)
         return;
 
-    tune_tcp_socket (fd, options.keepalive ? true : false);
-
     //  Create the engine object for this connection.
     stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options);
     alloc_assert (engine);
@@ -127,11 +124,7 @@ int xs::tcp_listener_t::set_address (const char *addr_)
         return -1;
 
     //  Create a listening socket.
-    s = open_socket (address.family (), SOCK_STREAM, IPPROTO_TCP);
-#ifdef XS_HAVE_WINDOWS
-    if (s == INVALID_SOCKET)
-        wsa_error_to_errno ();
-#endif
+    s = open_tcp_socket (address.family (), false);
 
     //  IPv6 address family not supported, try automatic downgrade to IPv4.
     if (address.family () == AF_INET6 && errno == EAFNOSUPPORT &&
@@ -139,18 +132,11 @@ int xs::tcp_listener_t::set_address (const char *addr_)
         rc = address.resolve (addr_, true, true);
         if (rc != 0)
             return rc;
-        s = ::socket (address.family (), SOCK_STREAM, IPPROTO_TCP);
+        s = open_tcp_socket (address.family (), false);
     }
 
-#ifdef XS_HAVE_WINDOWS
-    if (s == INVALID_SOCKET) {
-        wsa_error_to_errno ();
+    if (s == retired_fd)
         return -1;
-    }
-#else
-    if (s == -1)
-        return -1;
-#endif
 
     //  On some systems, IPv4 mapping in IPv6 sockets is disabled by default.
     //  Switch it on in such cases.
@@ -219,6 +205,7 @@ xs::fd_t xs::tcp_listener_t::accept ()
         return retired_fd;
     }
 #endif
+    tune_tcp_socket (sock, options.keepalive ? true : false);
     return sock;
 }