Set socket buffers before establishing TCP connection

The window scale option carried in SYN segment is computed from socket's
receive buffer size. So we need to set this buffer size before calling
connect or bind.

src/stream_engine.cpp

@@ -83,25 +83,6 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_,
     
     //  Put the socket into non-blocking mode.
     unblock_socket (s);
-    //  Set the socket buffer limits for the underlying socket.
-    if (options.sndbuf) {
-        rc = setsockopt (s, SOL_SOCKET, SO_SNDBUF,
-            (char*) &options.sndbuf, sizeof (int));
-#ifdef ZMQ_HAVE_WINDOWS
-		wsa_assert (rc != SOCKET_ERROR);
-#else
-        errno_assert (rc == 0);
-#endif
-    }
-    if (options.rcvbuf) {
-        rc = setsockopt (s, SOL_SOCKET, SO_RCVBUF,
-            (char*) &options.rcvbuf, sizeof (int));
-#ifdef ZMQ_HAVE_WINDOWS
-		wsa_assert (rc != SOCKET_ERROR);
-#else
-        errno_assert (rc == 0);
-#endif
-    }
 
 #ifdef SO_NOSIGPIPE
     //  Make sure that SIGPIPE signal is not generated when writing to a

src/tcp.cpp

@@ -59,6 +59,28 @@ void zmq::tune_tcp_socket (fd_t s_)
 #endif
 }
 
+void zmq::set_tcp_send_buffer (fd_t sockfd_, int bufsize_)
+{
+    const int rc = setsockopt (sockfd_, SOL_SOCKET, SO_SNDBUF,
+        (char*) &bufsize_, sizeof bufsize_);
+#ifdef ZMQ_HAVE_WINDOWS
+    wsa_assert (rc != SOCKET_ERROR);
+#else
+    errno_assert (rc == 0);
+#endif
+}
+
+void zmq::set_tcp_receive_buffer (fd_t sockfd_, int bufsize_)
+{
+    const int rc = setsockopt (sockfd_, SOL_SOCKET, SO_RCVBUF,
+        (char*) &bufsize_, sizeof bufsize_);
+#ifdef ZMQ_HAVE_WINDOWS
+    wsa_assert (rc != SOCKET_ERROR);
+#else
+    errno_assert (rc == 0);
+#endif
+}
+
 void zmq::tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_)
 {
     // These options are used only under certain #ifdefs below.

src/tcp.hpp

@@ -28,6 +28,12 @@ namespace zmq
     //  Tunes the supplied TCP socket for the best latency.
     void tune_tcp_socket (fd_t s_);
 
+    //  Sets the socket send buffer size.
+    void set_tcp_send_buffer (fd_t sockfd_, int bufsize_);
+
+    //  Sets the socket receive buffer size.
+    void set_tcp_receive_buffer (fd_t sockfd_, int bufsize_);
+
     //  Tunes TCP keep-alives
     void tune_tcp_keepalives (fd_t s_, int keepalive_, int keepalive_cnt_, int keepalive_idle_, int keepalive_intvl_);
 

src/tcp_connecter.cpp

@@ -226,6 +226,12 @@ int zmq::tcp_connecter_t::open ()
     // Set the socket to non-blocking mode so that we get async connect().
     unblock_socket (s);
 
+    //  Set the socket buffer limits for the underlying socket.
+    if (options.sndbuf != 0)
+        set_tcp_send_buffer (s, options.sndbuf);
+    if (options.rcvbuf != 0)
+        set_tcp_receive_buffer (s, options.rcvbuf);
+
     //  Connect to the remote peer.
     int rc = ::connect (
         s, addr->resolved.tcp_addr->addr (),

src/tcp_listener.cpp

@@ -188,6 +188,12 @@ int zmq::tcp_listener_t::set_address (const char *addr_)
     if (address.family () == AF_INET6)
         enable_ipv4_mapping (s);
 
+    //  Set the socket buffer limits for the underlying socket.
+    if (options.sndbuf != 0)
+        set_tcp_send_buffer (s, options.sndbuf);
+    if (options.rcvbuf != 0)
+        set_tcp_receive_buffer (s, options.rcvbuf);
+
     //  Allow reusing of the address.
     int flag = 1;
 #ifdef ZMQ_HAVE_WINDOWS