Skip to content

Commit

Permalink
initial implementation of tuning TCP keep-alives for TCP socket
Browse files Browse the repository at this point in the history
currently not fully cross-platform
work on linux
possibly work in *bsd
and could be enhanced to work on windows
  • Loading branch information
shripchenko committed Apr 5, 2012
1 parent 88db804 commit 0c3d917
Show file tree
Hide file tree
Showing 9 changed files with 223 additions and 3 deletions.
55 changes: 55 additions & 0 deletions acinclude.m4
Original file line number Diff line number Diff line change
Expand Up @@ -603,6 +603,61 @@ int main (int argc, char *argv [])
)
}])

dnl ################################################################################
dnl # LIBZMQ_CHECK_TCP_KEEPALIVE([action-if-found], [action-if-not-found]) #
dnl # Check if SO_KEEPALIVE is supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_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_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_TCP_KEEPALIVE_OPTS([action-if-found], [action-if-not-found]) #
dnl # Check if TCP_KEEPCNT, TCP_KEEPIDLE, TCP_KEEPINTVL are supported #
dnl ################################################################################
AC_DEFUN([LIBZMQ_CHECK_TCP_KEEPALIVE_OPTS], [{
AC_MSG_CHECKING(whether TCP_KEEPCNT TCP_KEEPIDLE TCP_KEEPINTVL are supported)
AC_TRY_RUN([/* TCP_KEEPCNT TCP_KEEPIDLE 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_KEEPCNT, (char*) &opt, sizeof (int))) == -1) ||
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPIDLE, (char*) &opt, sizeof (int))) == -1) ||
((rc = setsockopt (s, IPPROTO_TCP, TCP_KEEPINTVL, (char*) &opt, sizeof (int))) == -1)
);
}
],
[AC_MSG_RESULT(yes) ; libzmq_cv_tcp_keepalive_opts="yes" ; $1],
[AC_MSG_RESULT(no) ; libzmq_cv_tcp_keepalive_opts="no" ; $2],
[AC_MSG_RESULT(not during cross-compile) ; libzmq_cv_tcp_keepalive_opts="no"]
)
}])

dnl ################################################################################
dnl # LIBZMQ_CHECK_POLLER_KQUEUE([action-if-found], [action-if-not-found]) #
dnl # Checks kqueue polling system #
Expand Down
10 changes: 10 additions & 0 deletions configure.in
Original file line number Diff line number Diff line change
Expand Up @@ -383,6 +383,16 @@ LIBZMQ_CHECK_SOCK_CLOEXEC([AC_DEFINE(
[1],
[Whether SOCK_CLOEXEC is defined and functioning.])
])
LIBZMQ_CHECK_TCP_KEEPALIVE([AC_DEFINE(
[ZMQ_HAVE_TCP_KEEPALIVE],
[1],
[Whether SO_KEEPALIVE is supported.])
])
LIBZMQ_CHECK_TCP_KEEPALIVE_OPTS([AC_DEFINE(
[ZMQ_HAVE_TCP_KEEPALIVE_OPTS],
[1],
[Whether TCP_KEEPCNT, TCP_KEEPIDLE, TCP_KEEPINTVL are supported.])
])

# Subst LIBZMQ_EXTRA_CFLAGS & CXXFLAGS & LDFLAGS
AC_SUBST(LIBZMQ_EXTRA_CFLAGS)
Expand Down
5 changes: 5 additions & 0 deletions include/zmq.h
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
31 changes: 31 additions & 0 deletions src/ip.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,37 @@ 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_TCP_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

#if defined ZMQ_HAVE_TCP_KEEPALIVE_OPTS && !defined ZMQ_HAVE_WINDOWS
if (keepalive_cnt_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPCNT, &keepalive_cnt_, sizeof (int));
errno_assert (rc == 0);
}
if (keepalive_idle_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPIDLE, &keepalive_idle_, sizeof (int));
errno_assert (rc == 0);
}
if (keepalive_intvl_ != -1) {
int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPINTVL, &keepalive_intvl_, sizeof (int));
errno_assert (rc == 0);
}
#endif
}
#endif
}

void zmq::unblock_socket (fd_t s_)
{
#ifdef ZMQ_HAVE_WINDOWS
Expand Down
3 changes: 3 additions & 0 deletions src/ip.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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_);

Expand Down
111 changes: 109 additions & 2 deletions src/options.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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)
{
}
Expand Down Expand Up @@ -214,8 +218,75 @@ 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;
}
#ifdef ZMQ_HAVE_TCP_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_TCP_KEEPALIVE && defined ZMQ_HAVE_TCP_KEEPALIVE_OPTS
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_TCP_KEEPALIVE && defined ZMQ_HAVE_TCP_KEEPALIVE_OPTS
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_TCP_KEEPALIVE && defined ZMQ_HAVE_TCP_KEEPALIVE_OPTS
tcp_keepalive_intvl = val;
#endif
return 0;
}
}
errno = EINVAL;
return -1;
}
Expand Down Expand Up @@ -385,7 +456,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) {
Expand Down
9 changes: 8 additions & 1 deletion src/options.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
};
Expand Down
1 change: 1 addition & 0 deletions src/tcp_connecter.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
1 change: 1 addition & 0 deletions src/tcp_listener.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down

0 comments on commit 0c3d917

Please sign in to comment.