Merge pull request #1510 from pijyoi/retransmit_timeout

add ZMQ_TCP_RETRANSMIT_TIMEOUT socket option

doc/zmq_getsockopt.txt

@@ -672,6 +672,22 @@ Default value:: -1 (leave to OS default)
 Applicable socket types:: all, when using TCP transports.
 
 
+ZMQ_TCP_RETRANSMIT_TIMEOUT: Retrieve TCP Retransmit Timeout
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+On OSes where it is supported, retrieves how long before an unacknowledged TCP
+retransmit times out.
+The system normally attempts many TCP retransmits following an exponential
+backoff strategy. This means that after a network outage, it may take a long
+time before the session can be re-established. Setting this option allows
+the timeout to happen at a shorter interval.
+
+[horizontal]
+Option value type:: int
+Option value unit:: milliseconds
+Default value:: 0 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
 ZMQ_TOS: Retrieve the Type-of-Service socket override status
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Retrieve the IP_TOS option for the socket.

doc/zmq_setsockopt.txt

@@ -833,6 +833,22 @@ Default value:: -1 (leave to OS default)
 Applicable socket types:: all, when using TCP transports.
 
 
+ZMQ_TCP_RETRANSMIT_TIMEOUT: Set TCP Retransmit Timeout
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+On OSes where it is supported, sets how long before an unacknowledged TCP
+retransmit times out.
+The system normally attempts many TCP retransmits following an exponential
+backoff strategy. This means that after a network outage, it may take a long
+time before the session can be re-established. Setting this option allows
+the timeout to happen at a shorter interval.
+
+[horizontal]
+Option value type:: int
+Option value unit:: milliseconds
+Default value:: 0 (leave to OS default)
+Applicable socket types:: all, when using TCP transports.
+
+
 ZMQ_TOS: Set the Type-of-Service on socket
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Sets the ToS fields (Differentiated services (DS) and Explicit Congestion

include/zmq.h

@@ -321,6 +321,7 @@ ZMQ_EXPORT uint32_t zmq_msg_get_routing_id(zmq_msg_t *msg);
 #define ZMQ_HEARTBEAT_TIMEOUT 77
 #define ZMQ_XPUB_VERBOSE_UNSUBSCRIBE 78
 #define ZMQ_CONNECT_TIMEOUT 79
+#define ZMQ_TCP_RETRANSMIT_TIMEOUT 80
 
 /* Message options */
 #define ZMQ_MORE 1

src/options.cpp

@@ -47,6 +47,7 @@ zmq::options_t::options_t () :
  type (-1),
  linger (-1),
  connect_timeout (0),
+ tcp_retransmit_timeout (0),
  reconnect_ivl (100),
  reconnect_ivl_max (0),
  backlog (100),
@@ -166,6 +167,13 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
  }
  break;
 
+ case ZMQ_TCP_RETRANSMIT_TIMEOUT:
+ if (is_int && value >= 0) {
+ tcp_retransmit_timeout = value;
+ return 0;
+ }
+ break;
+
  case ZMQ_RECONNECT_IVL:
  if (is_int && value >= -1) {
  reconnect_ivl = value;
@@ -668,6 +676,13 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
  }
  break;
 
+ case ZMQ_TCP_RETRANSMIT_TIMEOUT:
+ if (is_int) {
+ *value = tcp_retransmit_timeout;
+ return 0;
+ }
+ break;
+
  case ZMQ_RECONNECT_IVL:
  if (is_int) {
  *value = reconnect_ivl;

src/options.hpp

@@ -97,6 +97,11 @@ namespace zmq
  // Default 0 (unused)
  int connect_timeout;
 
+ // Maximum interval in milliseconds beyond which TCP will timeout
+ // retransmitted packets.
+ // Default 0 (unused)
+ int tcp_retransmit_timeout;
+
  // Minimum interval between attempts to reconnect, in milliseconds.
  // Default 100ms
  int reconnect_ivl;

src/tcp.cpp

@@ -152,7 +152,25 @@ void zmq::tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int
 #endif // ZMQ_HAVE_WINDOWS
 }
 
-int zmq::tcp_write (fd_t s_, const void *data_, size_t size_)
+void zmq::tune_tcp_retransmit_timeout (fd_t sockfd_, int timeout_)
+{
+ if (timeout_ <= 0)
+ return;
+
+#if defined (ZMQ_HAVE_WINDOWS) && defined (TCP_MAXRT)
+ // msdn says it's supported in >= Vista, >= Windows Server 2003
+ timeout_ /= 1000; // in seconds
+ int rc = setsockopt (sockfd_, IPPROTO_TCP, TCP_MAXRT, (char*) &timeout_,
+ sizeof(timeout_));
+ wsa_assert (rc != SOCKET_ERROR);
+#elif defined (TCP_USER_TIMEOUT) // FIXME: should be ZMQ_HAVE_TCP_USER_TIMEOUT
+ int rc = setsockopt (sockfd_, IPPROTO_TCP, TCP_USER_TIMEOUT, &timeout_,
+ sizeof(timeout_));
+ errno_assert (rc == 0);
+#endif
+}
+
+ int zmq::tcp_write (fd_t s_, const void *data_, size_t size_)
 {
 #ifdef ZMQ_HAVE_WINDOWS
 

src/tcp.hpp

@@ -47,6 +47,9 @@ namespace zmq
  // Tunes TCP keep-alives
  void tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_);
 
+ // Tunes TCP retransmit timeout
+ void tune_tcp_retransmit_timeout (fd_t sockfd_, int timeout_);
+
  // Writes data to the socket. Returns the number of bytes actually
  // written (even zero is to be considered to be a success). In case
  // of error or orderly shutdown by the other peer -1 is returned.

src/tcp_connecter.cpp

@@ -145,6 +145,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);
+ tune_tcp_retransmit_timeout (fd, options.tcp_retransmit_timeout);
 
  // remember our fd for ZMQ_SRCFD in messages
  socket->set_fd (fd);