Merge pull request #306 from shripchenko/master

Implement ZMQ_TCP_KEEPALIVE* family of options to adjust TCP keepalives

acinclude.m4

@@ -603,6 +603,143 @@ int main (int argc, char *argv [])
     )
 }])
 
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_SO_KEEPALIVE([action-if-found], [action-if-not-found])          #
+dnl # Check if SO_KEEPALIVE is supported                                           #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_SO_KEEPALIVE], [{
+    AC_MSG_CHECKING(whether SO_KEEPALIVE is supported)
+    AC_TRY_RUN([/* SO_KEEPALIVE test */
+#include <sys/types.h>
+#include <sys/socket.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_so_keepalive="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_so_keepalive="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_so_keepalive="no"]
+    )
+}])
+
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_TCP_KEEPCNT([action-if-found], [action-if-not-found])           #
+dnl # Check if TCP_KEEPCNT is supported                                            #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPCNT], [{
+    AC_MSG_CHECKING(whether TCP_KEEPCNT is supported)
+    AC_TRY_RUN([/* TCP_KEEPCNT test */
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1) ||
+        ((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPCNT, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepcnt="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_tcp_keepcnt="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepcnt="no"]
+    )
+}])
+
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_TCP_KEEPIDLE([action-if-found], [action-if-not-found])          #
+dnl # Check if TCP_KEEPIDLE is supported                                           #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPIDLE], [{
+    AC_MSG_CHECKING(whether TCP_KEEPIDLE is supported)
+    AC_TRY_RUN([/* TCP_KEEPIDLE test */
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1) ||
+        ((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPIDLE, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepidle="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_tcp_keepidle="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepidle="no"]
+    )
+}])
+
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_TCP_KEEPINTVL([action-if-found], [action-if-not-found])          #
+dnl # Check if TCP_KEEPINTVL is supported                                           #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPINTVL], [{
+    AC_MSG_CHECKING(whether TCP_KEEPINTVL is supported)
+    AC_TRY_RUN([/* TCP_KEEPINTVL test */
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1) ||
+        ((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPINTVL, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepintvl="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_tcp_keepintvl="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepintvl="no"]
+    )
+}])
+
+dnl ################################################################################
+dnl # LIBZMQ_CHECK_TCP_KEEPALIVE([action-if-found], [action-if-not-found])         #
+dnl # Check if TCP_KEEPALIVE is supported                                          #
+dnl ################################################################################
+AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPALIVE], [{
+    AC_MSG_CHECKING(whether TCP_KEEPALIVE is supported)
+    AC_TRY_RUN([/* TCP_KEEPALIVE test */
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+
+int main (int argc, char *argv [])
+{
+    int s, rc, opt = 1;
+    return (
+        ((s = socket (PF_INET, SOCK_STREAM, 0)) == -1) ||
+        ((rc = setsockopt (s, SOL_SOCKET, SO_KEEPALIVE, (char*) &opt, sizeof (int))) == -1) ||
+        ((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPALIVE, (char*) &opt, sizeof (int))) == -1)
+    );
+}
+    ],
+    [AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepalive="yes" ; $1],
+    [AC_MSG_RESULT(no)  ; libzmq_cv_tcp_keepalive="no"  ; $2],
+    [AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepalive="no"]
+    )
+}])
+
 dnl ################################################################################
 dnl # LIBZMQ_CHECK_POLLER_KQUEUE([action-if-found], [action-if-not-found])         #
 dnl # Checks kqueue polling system                                                 #

configure.in

@@ -384,6 +384,33 @@ LIBZMQ_CHECK_SOCK_CLOEXEC([AC_DEFINE(
                               [Whether SOCK_CLOEXEC is defined and functioning.])
                           ])
 
+# TCP keep-alives Checks.
+LIBZMQ_CHECK_SO_KEEPALIVE([AC_DEFINE(
+                              [ZMQ_HAVE_SO_KEEPALIVE],
+                              [1],
+                              [Whether SO_KEEPALIVE is supported.])
+                          ])
+LIBZMQ_CHECK_TCP_KEEPCNT([AC_DEFINE(
+                              [ZMQ_HAVE_TCP_KEEPCNT],
+                              [1],
+                              [Whether TCP_KEEPCNT is supported.])
+                          ])
+LIBZMQ_CHECK_TCP_KEEPIDLE([AC_DEFINE(
+                              [ZMQ_HAVE_TCP_KEEPIDLE],
+                              [1],
+                              [Whether TCP_KEEPIDLE is supported.])
+                          ])
+LIBZMQ_CHECK_TCP_KEEPINTVL([AC_DEFINE(
+                              [ZMQ_HAVE_TCP_KEEPINTVL],
+                              [1],
+                              [Whether TCP_KEEPINTVL is supported.])
+                          ])
+LIBZMQ_CHECK_TCP_KEEPALIVE([AC_DEFINE(
+                              [ZMQ_HAVE_TCP_KEEPALIVE],
+                              [1],
+                              [Whether TCP_KEEPALIVE is supported.])
+                          ])
+
 # Subst LIBZMQ_EXTRA_CFLAGS & CXXFLAGS & LDFLAGS
 AC_SUBST(LIBZMQ_EXTRA_CFLAGS)
 AC_SUBST(LIBZMQ_EXTRA_CXXFLAGS)

doc/zmq_getsockopt.txt

@@ -404,6 +404,51 @@ Option value unit:: N/A
 Default value:: NULL
 Applicable socket types:: all, when binding TCP or IPC transports
 
+ZMQ_TCP_KEEPALIVE: Override SO_KEEPALIVE socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'SO_KEEPALIVE' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,0,1
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+ZMQ_TCP_KEEPALIVE_IDLE: Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPCNT'(or 'TCP_KEEPALIVE' on some OS) socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+ZMQ_TCP_KEEPALIVE_CNT: Override TCP_KEEPCNT socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPCNT' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+ZMQ_TCP_KEEPALIVE_INTVL: Override TCP_KEEPINTVL socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPINTVL' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
 RETURN VALUE
 ------------
 The _zmq_getsockopt()_ function shall return zero if successful. Otherwise it

doc/zmq_setsockopt.txt

@@ -367,6 +367,54 @@ Default value:: 0 (false)
 Applicable socket types:: ZMQ_ROUTER
 
 
+ZMQ_TCP_KEEPALIVE: Override SO_KEEPALIVE socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'SO_KEEPALIVE' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,0,1
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
+ZMQ_TCP_KEEPALIVE_IDLE: Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPCNT'(or 'TCP_KEEPALIVE' on some OS) socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
+ZMQ_TCP_KEEPALIVE_CNT: Override TCP_KEEPCNT socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPCNT' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
+ZMQ_TCP_KEEPALIVE_INTVL: Override TCP_KEEPINTVL socket option
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Override 'TCP_KEEPINTVL' socket option(where supported by OS).
+The default value of `-1` means to skip any overrides and leave it to OS default.
+
+[horizontal]
+Option value type:: int
+Option value unit:: -1,>0
+Default value:: -1 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
 RETURN VALUE
 ------------
 The _zmq_setsockopt()_ function shall return zero if successful. Otherwise it

include/zmq.h

@@ -222,6 +222,11 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
 #define ZMQ_IPV4ONLY 31
 #define ZMQ_LAST_ENDPOINT 32
 #define ZMQ_FAIL_UNROUTABLE 33
+#define ZMQ_TCP_KEEPALIVE 34
+#define ZMQ_TCP_KEEPALIVE_CNT 35
+#define ZMQ_TCP_KEEPALIVE_IDLE 36
+#define ZMQ_TCP_KEEPALIVE_INTVL 37
+
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1

src/ip.cpp

@@ -83,6 +83,66 @@ void zmq::tune_tcp_socket (fd_t s_)
 #endif
 }
 
+void zmq::tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_)
+{
+    //  Tuning TCP keep-alives if platform allows it
+    //  All values = -1 means skip and leave it for OS
+#ifdef ZMQ_HAVE_SO_KEEPALIVE
+    if (keepalive_ != -1) {
+        int rc = setsockopt (s_, SOL_SOCKET, SO_KEEPALIVE, (char*) &keepalive_, sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+        wsa_assert (rc != SOCKET_ERROR);
+#else
+        errno_assert (rc == 0);
+#endif
+
+#ifdef ZMQ_HAVE_TCP_KEEPCNT
+        if (keepalive_cnt_ != -1) {
+            int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPCNT, &keepalive_cnt_, sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+            wsa_assert (rc != SOCKET_ERROR);
+#else
+            errno_assert (rc == 0);
+#endif
+        }
+#endif // ZMQ_HAVE_TCP_KEEPCNT
+
+#ifdef ZMQ_HAVE_TCP_KEEPIDLE
+        if (keepalive_idle_ != -1) {
+            int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPIDLE, &keepalive_idle_, sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+            wsa_assert (rc != SOCKET_ERROR);
+#else
+            errno_assert (rc == 0);
+#endif
+        }
+#else // ZMQ_HAVE_TCP_KEEPIDLE
+#ifdef ZMQ_HAVE_TCP_KEEPALIVE
+        if (keepalive_idle_ != -1) {
+            int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPALIVE, &keepalive_idle_, sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+            wsa_assert (rc != SOCKET_ERROR);
+#else
+            errno_assert (rc == 0);
+#endif
+        }
+#endif // ZMQ_HAVE_TCP_KEEPALIVE
+#endif // ZMQ_HAVE_TCP_KEEPIDLE
+
+#ifdef ZMQ_HAVE_TCP_KEEPINTVL
+        if (keepalive_intvl_ != -1) {
+            int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPINTVL, &keepalive_intvl_, sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+            wsa_assert (rc != SOCKET_ERROR);
+#else
+            errno_assert (rc == 0);
+#endif
+        }
+#endif // ZMQ_HAVE_TCP_KEEPINTVL
+    }
+#endif // ZMQ_HAVE_SO_KEEPALIVE
+}
+
 void zmq::unblock_socket (fd_t s_)
 {
 #ifdef ZMQ_HAVE_WINDOWS

src/ip.hpp

@@ -33,6 +33,9 @@ namespace zmq
     //  Tunes the supplied TCP socket for the best latency.
     void tune_tcp_socket (fd_t s_);
 
+    //  Tunes TCP keep-alives
+    void tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_);
+
     //  Sets the socket into non-blocking mode.
     void unblock_socket (fd_t s_);
 

src/options.cpp

@@ -49,6 +49,10 @@ zmq::options_t::options_t () :
     filter (false),
     send_identity (false),
     recv_identity (false),
+    tcp_keepalive (-1),
+    tcp_keepalive_cnt (-1),
+    tcp_keepalive_idle (-1),
+    tcp_keepalive_intvl (-1),
     socket_id (0)
 {
 }
@@ -214,8 +218,76 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
             ipv4only = val;
             return 0;
         }
-    }
 
+    case ZMQ_TCP_KEEPALIVE:
+        {
+            if (optvallen_ != sizeof (int)) {
+                errno = EINVAL;
+                return -1;
+            }
+            int val = *((int*) optval_);
+            if (val != -1 && val != 0 && val != 1) {
+                errno = EINVAL;
+                return -1;
+            }
+#if defined ZMQ_HAVE_SO_KEEPALIVE
+            tcp_keepalive = val;
+#endif
+            return 0;
+        }
+
+    case ZMQ_TCP_KEEPALIVE_CNT:
+        {
+            if (optvallen_ != sizeof (int)) {
+                errno = EINVAL;
+                return -1;
+            }
+            int val = *((int*) optval_);
+            if (val <= 0 && val != -1) {
+                errno = EINVAL;
+                return -1;
+            }
+#if defined ZMQ_HAVE_SO_KEEPALIVE && defined ZMQ_HAVE_TCP_KEEPCNT
+            tcp_keepalive_cnt = val;
+#endif
+            return 0;
+        }
+
+    case ZMQ_TCP_KEEPALIVE_IDLE:
+        {
+            if (optvallen_ != sizeof (int)) {
+                errno = EINVAL;
+                return -1;
+            }
+            int val = *((int*) optval_);
+            if (val <= 0 && val != -1) {
+                errno = EINVAL;
+                return -1;
+            }
+
+#if defined ZMQ_HAVE_SO_KEEPALIVE && (defined ZMQ_HAVE_TCP_KEEPIDLE || defined ZMQ_HAVE_TCP_KEEPALIVE)
+            tcp_keepalive_idle = val;
+#endif
+            return 0;
+        }
+
+    case ZMQ_TCP_KEEPALIVE_INTVL:
+        {
+            if (optvallen_ != sizeof (int)) {
+                errno = EINVAL;
+                return -1;
+            }
+            int val = *((int*) optval_);
+            if (val <= 0 && val != -1) {
+                errno = EINVAL;
+                return -1;
+            }
+#if defined ZMQ_HAVE_SO_KEEPALIVE && defined ZMQ_HAVE_TCP_KEEPINTVL
+            tcp_keepalive_intvl = val;
+#endif
+            return 0;
+        }
+    }
     errno = EINVAL;
     return -1;
 }
@@ -385,7 +457,43 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
         *((int*) optval_) = ipv4only;
         *optvallen_ = sizeof (int);
         return 0;
-        
+
+    case ZMQ_TCP_KEEPALIVE:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = tcp_keepalive;
+        *optvallen_ = sizeof (int);
+        return 0;
+
+    case ZMQ_TCP_KEEPALIVE_CNT:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = tcp_keepalive_cnt;
+        *optvallen_ = sizeof (int);
+        return 0;
+
+    case ZMQ_TCP_KEEPALIVE_IDLE:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = tcp_keepalive_idle;
+        *optvallen_ = sizeof (int);
+        return 0;
+
+    case ZMQ_TCP_KEEPALIVE_INTVL:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = tcp_keepalive_intvl;
+        *optvallen_ = sizeof (int);
+        return 0;
+
     case ZMQ_LAST_ENDPOINT:
         // don't allow string which cannot contain the entire message
         if (*optvallen_ < last_endpoint.size() + 1) {

src/options.hpp

@@ -110,7 +110,14 @@ namespace zmq
 
         //  Receivers identity from all new connections.
         bool recv_identity;
-        
+
+        //  TCP keep-alive settings.
+        //  Defaults to -1 = do not change socket options
+        int tcp_keepalive;
+        int tcp_keepalive_cnt;
+        int tcp_keepalive_idle;
+        int tcp_keepalive_intvl;
+
         //  ID of the socket.
         int socket_id;
     };

src/tcp_connecter.cpp

@@ -106,6 +106,7 @@ void zmq::tcp_connecter_t::out_event ()
     }
 
     tune_tcp_socket (fd);
+    tune_tcp_keepalives (fd, options.tcp_keepalive, options.tcp_keepalive_cnt, options.tcp_keepalive_idle, options.tcp_keepalive_intvl);
 
     //  Create the engine object for this connection.
     stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options);

src/tcp_listener.cpp

@@ -87,6 +87,7 @@ void zmq::tcp_listener_t::in_event ()
         return;
 
     tune_tcp_socket (fd);
+    tune_tcp_keepalives (fd, options.tcp_keepalive, options.tcp_keepalive_cnt, options.tcp_keepalive_idle, options.tcp_keepalive_intvl);
 
     //  Create the engine object for this connection.
     stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options);