XS_KEEPALIVE options added

This option allows to turn on TCP keepalives on the underlying connections.

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

6 files changed, 43 insertions, 4 deletions

diff --git a/src/ip.cpp b/src/ip.cpp
index 1eca352..bae8d7e 100644
--- a/src/ip.cpp
+++ b/src/ip.cpp
@@ -60,7 +60,7 @@ xs::fd_t xs::open_socket (int domain_, int type_, int protocol_)
     return s;
 }
 
-void xs::tune_tcp_socket (fd_t s_)
+void xs::tune_tcp_socket (fd_t s_, bool keepalive_)
 {
     //  Disable Nagle's algorithm. We are doing data batching on Crossroads
     //  level, so using Nagle wouldn't improve throughput in anyway, but it
@@ -81,6 +81,17 @@ void xs::tune_tcp_socket (fd_t s_)
         sizeof (int));
     errno_assert (rc != SOCKET_ERROR);
 #endif
+
+    if (keepalive_) { 
+        int keepalive = 1;
+        int rc = setsockopt (s_, SOL_SOCKET, SO_KEEPALIVE, (char*) &keepalive,
+            sizeof (int));
+#ifdef XS_HAVE_WINDOWS
+        wsa_assert (rc != SOCKET_ERROR);
+#else
+        errno_assert (rc == 0);
+#endif
+    }
 }
 
 void xs::unblock_socket (fd_t s_)
diff --git a/src/ip.hpp b/src/ip.hpp
index fabe953..9487303 100644
--- a/src/ip.hpp
+++ b/src/ip.hpp
@@ -31,7 +31,7 @@ namespace xs
     fd_t open_socket (int domain_, int type_, int protocol_);
 
     //  Tunes the supplied TCP socket for the best latency.
-    void tune_tcp_socket (fd_t s_);
+    void tune_tcp_socket (fd_t s_, bool keeaplive_);
 
     //  Sets the socket into non-blocking mode.
     void unblock_socket (fd_t s_);
diff --git a/src/options.cpp b/src/options.cpp
index 2b123c2..07d3752 100644
--- a/src/options.cpp
+++ b/src/options.cpp
@@ -45,6 +45,7 @@ xs::options_t::options_t () :
     rcvtimeo (-1),
     sndtimeo (-1),
     ipv4only (1),
+    keepalive (0),
     delay_on_close (true),
     delay_on_disconnect (true),
     filter (false),
@@ -216,6 +217,21 @@ int xs::options_t::setsockopt (int option_, const void *optval_,
             return 0;
         }
 
+    case XS_KEEPALIVE:
+        {
+            if (optvallen_ != sizeof (int)) {
+                errno = EINVAL;
+                return -1;
+            }
+            int val = *((int*) optval_);
+            if (val != 0 && val != 1) {
+                errno = EINVAL;
+                return -1;
+            }
+            keepalive = val;
+            return 0;
+        }
+
     }
 
     errno = EINVAL;
@@ -388,6 +404,15 @@ int xs::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
         *optvallen_ = sizeof (int);
         return 0;
 
+    case XS_KEEPALIVE:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = keepalive;
+        *optvallen_ = sizeof (int);
+        return 0;
+
     }
 
     errno = EINVAL;
diff --git a/src/options.hpp b/src/options.hpp
index 1ea61bd..3e47336 100644
--- a/src/options.hpp
+++ b/src/options.hpp
@@ -89,6 +89,9 @@ namespace xs
         //  connect to and accept connections from both IPv4 and IPv6 hosts.
         int ipv4only;
 
+        //  If 1, keepalives are to be sent periodically.
+        int keepalive;
+
         //  If true, session reads all the pending messages from the pipe and
         //  sends them to the network when socket is closed.
         bool delay_on_close;
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index f938967..c819f86 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -108,7 +108,7 @@ void xs::tcp_connecter_t::out_event (fd_t fd_)
         return;
     }
 
-    tune_tcp_socket (fd);
+    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);
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index 562fad5..c0ea564 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -86,7 +86,7 @@ void xs::tcp_listener_t::in_event (fd_t fd_)
     if (fd == retired_fd)
         return;
 
-    tune_tcp_socket (fd);
+    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);