Merge branch 'master' of github.com:ianbarber/libzmq

src/Makefile.am

@@ -65,6 +65,7 @@ libzmq_la_SOURCES = \
     stdint.hpp \
     stream_engine.hpp \
     sub.hpp \
+    tcp.hpp \
     tcp_address.hpp \
     tcp_connecter.hpp \
     tcp_listener.hpp \
@@ -123,6 +124,7 @@ libzmq_la_SOURCES = \
     socket_base.cpp \
     stream_engine.cpp \
     sub.cpp \
+    tcp.cpp \
     tcp_address.cpp \
     tcp_connecter.cpp \
     tcp_listener.cpp \

src/decoder.cpp

@@ -145,7 +145,7 @@ bool zmq::decoder_t::eight_byte_size_ready ()
 bool zmq::decoder_t::flags_ready ()
 {
     //  Store the flags from the wire into the message structure.
-    in_progress.set_flags (tmpbuf [0]);
+    in_progress.set_flags (tmpbuf [0] & msg_t::more);
 
     next_step (in_progress.data (), in_progress.size (),
         &decoder_t::message_ready);

src/fq.cpp

@@ -84,11 +84,6 @@ int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
         //  subsequent part should be immediately available.
         bool fetched = pipes [current]->read (msg_);
 
-        //  Check the atomicity of the message. If we've already received the
-        //  first part of the message we should get the remaining parts
-        //  without blocking.
-        zmq_assert (!more || fetched);
-
         //  Note that when message is not fetched, current pipe is deactivated
         //  and replaced by another active pipe. Thus we don't have to increase
         //  the 'current' pointer.
@@ -101,6 +96,11 @@ int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
             return 0;
         }
 
+        //  Check the atomicity of the message.
+        //  If we've already received the first part of the message
+        //  we should get the remaining parts without blocking.
+        zmq_assert (!more);
+
         active--;
         pipes.swap (current, active);
         if (current == active)

src/ip.cpp

@@ -66,89 +66,6 @@ zmq::fd_t zmq::open_socket (int domain_, int type_, int protocol_)
     return s;
 }
 
-void zmq::tune_tcp_socket (fd_t s_)
-{
-    //  Disable Nagle's algorithm. We are doing data batching on 0MQ level,
-    //  so using Nagle wouldn't improve throughput in anyway, but it would
-    //  hurt latency.
-    int nodelay = 1;
-    int rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay,
-        sizeof (int));
-#ifdef ZMQ_HAVE_WINDOWS
-    wsa_assert (rc != SOCKET_ERROR);
-#else
-    errno_assert (rc == 0);
-#endif
-
-#ifdef ZMQ_HAVE_OPENVMS
-    //  Disable delayed acknowledgements as they hurt latency is serious manner.
-    int nodelack = 1;
-    rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack,
-        sizeof (int));
-    errno_assert (rc != SOCKET_ERROR);
-#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

@@ -30,12 +30,6 @@ namespace zmq
     //  Same as socket(2), but allows for transparent tweaking the options.
     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_);
-
-    //  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/session_base.cpp

@@ -111,6 +111,7 @@ zmq::session_base_t::session_base_t (class io_thread_t *io_thread_,
     io_object_t (io_thread_),
     connect (connect_),
     pipe (NULL),
+    incomplete_detach (0),
     incomplete_in (false),
     pending (false),
     engine (NULL),
@@ -229,8 +230,19 @@ void zmq::session_base_t::clean_pipes ()
 
 void zmq::session_base_t::terminated (pipe_t *pipe_)
 {
-    //  Drop the reference to the deallocated pipe.
-    zmq_assert (pipe == pipe_);
+    //  Drop the reference to the deallocated pipe if required.
+    zmq_assert (pipe == pipe_ || incomplete_detach > 0);
+    
+    //  If we still have pipes outstanding, decrement.
+    //  This will only have been set in a disconnect situation
+    //  where delay_attach_on_connect is 1. 
+    if (incomplete_detach > 0)
+        incomplete_detach --;
+
+    //  If there are still extra detached pipes, don't continue
+    if (incomplete_detach > 0)
+        return;
+    
     pipe = NULL;
 
     //  If we are waiting for pending messages to be sent, at this point
@@ -242,6 +254,10 @@ void zmq::session_base_t::terminated (pipe_t *pipe_)
 
 void zmq::session_base_t::read_activated (pipe_t *pipe_)
 {
+    //  Skip activating if we're detaching this pipe
+    if (incomplete_detach > 0 && pipe_ != pipe) 
+        return;
+        
     zmq_assert (pipe == pipe_);
 
     if (likely (engine != NULL))
@@ -252,6 +268,10 @@ void zmq::session_base_t::read_activated (pipe_t *pipe_)
 
 void zmq::session_base_t::write_activated (pipe_t *pipe_)
 {
+    //  Skip activating if we're detaching this pipe
+    if (incomplete_detach > 0 && pipe_ != pipe) 
+        return;
+        
     zmq_assert (pipe == pipe_);
 
     if (engine)
@@ -291,7 +311,7 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
     zmq_assert (engine_ != NULL);
 
     //  Create the pipe if it does not exist yet.
-    if (!pipe && !is_terminating ()) {
+    if ((!pipe || incomplete_detach > 0) && !is_terminating ()) {
         object_t *parents [2] = {this, socket};
         pipe_t *pipes [2] = {NULL, NULL};
         int hwms [2] = {options.rcvhwm, options.sndhwm};
@@ -308,9 +328,6 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
 
         //  Ask socket to plug into the remote end of the pipe.
         send_bind (socket, pipes [1]);
-
-        //  Store the outpipe for disconnect situations
-        outpipe = pipes [1];
     }
 
     //  Plug in the engine.
@@ -398,8 +415,17 @@ void zmq::session_base_t::detached ()
         return;
     }
 
-    reset ();
+    //  For delayed connect situations, terminate the pipe
+    //  and reestablish later on
+    if (pipe && options.delay_attach_on_connect == 1
+        && addr->protocol != "pgm" && addr->protocol != "epgm") {
+        pipe->hiccup ();
+        pipe->terminate (false);
+        incomplete_detach ++;
+    }
 
+    reset ();
+    
     //  Reconnect.
     if (options.reconnect_ivl != -1)
         start_connecting (true);
@@ -408,14 +434,6 @@ void zmq::session_base_t::detached ()
     //  the socket object to resend all the subscriptions.
     if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB))
         pipe->hiccup ();
-
-    //  For delayed connect situations, terminate the pipe
-    //  and reestablish later on
-    if (pipe && options.delay_attach_on_connect == 1
-        && addr->protocol != "pgm" && addr->protocol != "epgm") {
-        pipe->terminate (false);
-        outpipe->terminate (false);
-    }
 }
 
 void zmq::session_base_t::start_connecting (bool wait_)

src/session_base.hpp

@@ -104,8 +104,8 @@ namespace zmq
         //  Pipe connecting the session to its socket.
         zmq::pipe_t *pipe;
         
-        //  Socket end of the pipe, for delay attach scenario
-        zmq::pipe_t *outpipe;
+        //  This flag is set if we are disconnecting, but haven't yet completed
+        int incomplete_detach;
 
         //  This flag is true if the remainder of the message being processed
         //  is still in the in pipe.

src/signaler.cpp

@@ -298,7 +298,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     wsa_assert (rc != SOCKET_ERROR);
 
     //  Connect writer to the listener.
-    rc = connect (*w_, (sockaddr *) &addr, sizeof (addr));
+    rc = connect (*w_, (struct sockaddr*) &addr, sizeof (addr));
     wsa_assert (rc != SOCKET_ERROR);
 
     //  Accept connection from writer.
@@ -327,7 +327,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     //
     //  The bug will be fixed in V5.6 ECO4 and beyond.  In the meantime, we'll
     //  create the socket pair manually.
-    sockaddr_in lcladdr;
+    struct sockaddr_in lcladdr;
     memset (&lcladdr, 0, sizeof (lcladdr));
     lcladdr.sin_family = AF_INET;
     lcladdr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);

src/socket_base.cpp

@@ -970,7 +970,11 @@ void zmq::socket_base_t::write_activated (pipe_t *pipe_)
 
 void zmq::socket_base_t::hiccuped (pipe_t *pipe_)
 {
-    xhiccuped (pipe_);
+    if (options.delay_attach_on_connect == 1) 
+        pipe_->terminate (false);
+    else 
+        //  Notify derived sockets of the hiccup
+        xhiccuped (pipe_);
 }
 
 void zmq::socket_base_t::terminated (pipe_t *pipe_)

src/tcp.cpp

+/*
+    Copyright (c) 2010-2011 250bpm s.r.o.
+    Copyright (c) 2007-2009 iMatix Corporation
+    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "ip.hpp"
+#include "tcp.hpp"
+#include "err.hpp"
+#include "platform.hpp"
+
+#if defined ZMQ_HAVE_WINDOWS
+#include "windows.hpp"
+#else
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#endif
+
+#if defined ZMQ_HAVE_OPENVMS
+#include <ioctl.h>
+#endif
+
+void zmq::tune_tcp_socket (fd_t s_)
+{
+    //  Disable Nagle's algorithm. We are doing data batching on 0MQ level,
+    //  so using Nagle wouldn't improve throughput in anyway, but it would
+    //  hurt latency.
+    int nodelay = 1;
+    int rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay,
+        sizeof (int));
+#ifdef ZMQ_HAVE_WINDOWS
+    wsa_assert (rc != SOCKET_ERROR);
+#else
+    errno_assert (rc == 0);
+#endif
+
+#ifdef ZMQ_HAVE_OPENVMS
+    //  Disable delayed acknowledgements as they hurt latency is serious manner.
+    int nodelack = 1;
+    rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack,
+        sizeof (int));
+    errno_assert (rc != SOCKET_ERROR);
+#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
+}

src/tcp.hpp

+/*
+    Copyright (c) 2010-2011 250bpm s.r.o.
+    Copyright (c) 2007-2009 iMatix Corporation
+    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_TCP_HPP_INCLUDED__
+#define __ZMQ_TCP_HPP_INCLUDED__
+
+#include "fd.hpp"
+
+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_);
+
+}
+
+#endif 

src/tcp_connecter.cpp

@@ -29,6 +29,7 @@
 #include "random.hpp"
 #include "err.hpp"
 #include "ip.hpp"
+#include "tcp.hpp"
 #include "address.hpp"
 #include "tcp_address.hpp"
 #include "session_base.hpp"

src/tcp_listener.cpp

@@ -31,6 +31,7 @@
 #include "config.hpp"
 #include "err.hpp"
 #include "ip.hpp"
+#include "tcp.hpp"
 #include "socket_base.hpp"
 
 #ifdef ZMQ_HAVE_WINDOWS

tests/test_connect_delay.cpp

@@ -27,6 +27,111 @@
 
 #include "../include/zmq.h"
 
+static void *server (void *c)
+{
+    void *socket, *context;
+    char buffer[16];
+    int rc, val;
+
+    shoulddie = *(long *)sd;
+
+    context = zmq_init (1);
+    assert (context);
+
+    socket = zmq_socket (context, ZMQ_PULL);
+    assert (socket);
+
+    val = 0;
+    rc = zmq_setsockopt(socket, ZMQ_LINGER, &val, sizeof(val));
+    assert (rc == 0);
+
+    rc = zmq_bind (socket, "ipc:///tmp/recon");
+    assert (rc == 0);
+
+    memset (&buffer, 0, sizeof(buffer));
+    rc = zmq_recv (socket, &buffer, sizeof(buffer), 0);
+
+    //  Intentionally bail out
+    rc = zmq_close (socket);
+    assert (rc == 0);
+
+    rc = zmq_term (context);
+    assert (rc == 0);
+
+    usleep (200000);
+
+    context = zmq_init (1);
+    assert (context);
+
+    socket = zmq_socket (context, ZMQ_PULL);
+    assert (socket);
+
+    val = 0;
+    rc = zmq_setsockopt(socket, ZMQ_LINGER, &val, sizeof(val));
+    assert (rc == 0);
+
+    rc = zmq_bind (socket, "ipc:///tmp/recon");
+    assert (rc == 0);
+
+    usleep (200000);
+
+    memset (&buffer, 0, sizeof(buffer));
+    rc = zmq_recv (socket, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
+    assert (rc != -1);
+
+    //  Start closing the socket while the connecting process is underway.
+    rc = zmq_close (socket);
+    assert (rc == 0);
+
+    rc = zmq_term (context);
+    assert (rc == 0);
+
+    pthread_exit(NULL);
+}
+
+static void *worker (void *n)
+{
+    void *socket, *context;
+    int rc, hadone, val;
+
+    context = zmq_init (1);
+    assert (context);
+
+    socket = zmq_socket (context, ZMQ_PUSH);
+    assert (socket);
+
+    val = 0;
+    rc = zmq_setsockopt(socket, ZMQ_LINGER, &val, sizeof(val));
+    assert (rc == 0);
+
+    val = 1;
+    rc = zmq_setsockopt (socket, ZMQ_DELAY_ATTACH_ON_CONNECT, &val, sizeof(val));
+    assert (rc == 0);
+
+    rc = zmq_connect (socket, "ipc:///tmp/recon");
+    assert (rc == 0);
+
+    hadone = 0;
+    //  Not checking RC as some may be -1
+    for (int i = 0; i < 4; i++) {
+        usleep(200000);
+        rc = zmq_send (socket, "hi", 2, ZMQ_DONTWAIT);
+        if (rc != -1) 
+            hadone ++;
+    }
+
+    assert (hadone >= 2);
+    assert (hadone < 4);
+
+    rc = zmq_close (socket);
+    assert (rc == 0);
+
+    rc = zmq_term (context);
+    assert (rc == 0);
+
+    pthread_exit(NULL);
+}
+
 int main (int argc, char *argv [])
 {
     fprintf (stderr, "test_connect_delay running...\n");
@@ -69,7 +174,7 @@ int main (int argc, char *argv [])
     seen = 0;
     for (int i = 0; i < 10; ++i)
     {
-        memset(&buffer, 0, sizeof(buffer));
+        memset (&buffer, 0, sizeof(buffer));
         rc = zmq_recv (to, &buffer, sizeof(buffer), ZMQ_DONTWAIT);
         if( rc == -1) 
             break;
@@ -139,9 +244,19 @@ int main (int argc, char *argv [])
 
     rc = zmq_close (to);
     assert (rc == 0);
-
+    
     rc = zmq_ctx_destroy(context);
     assert (rc == 0);
 
-    return 0;
+    fprintf (stderr, " Running DELAY_ATTACH_ON_CONNECT with disconnect\n");
+
+    pthread_t serv, work;
+
+    rc = pthread_create (&serv, NULL, server, NULL);
+    assert (rc == 0);
+
+    rc = pthread_create (&work, NULL, worker, NULL);
+    assert (rc == 0);
+    
+    pthread_exit(NULL);
 }

tests/test_connect_resolve.cpp

@@ -40,7 +40,7 @@ int main (int argc, char *argv [])
     int rc = zmq_connect (sock, "tcp://localhost:1234");
     assert (rc == 0);
 
-    rc = zmq_connect (sock, "tcp://foobar123xyz:1234");
+    rc = zmq_connect (sock, "tcp://0mq.is.teh.best:1234");
     assert (rc == -1);
     assert (errno == EINVAL);