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

This reverts commit 33459029, reversing
changes made to 889b0e6f.

src/Makefile.am

@@ -65,7 +65,6 @@ libzmq_la_SOURCES = \
     stdint.hpp \
     stream_engine.hpp \
     sub.hpp \
-    tcp.hpp \
     tcp_address.hpp \
     tcp_connecter.hpp \
     tcp_listener.hpp \
@@ -124,7 +123,6 @@ 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] & msg_t::more);
+    in_progress.set_flags (tmpbuf [0]);
 
     next_step (in_progress.data (), in_progress.size (),
         &decoder_t::message_ready);

src/fq.cpp

@@ -84,6 +84,11 @@ 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.
@@ -96,11 +101,6 @@ 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,6 +66,89 @@ 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,6 +30,12 @@ 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,7 +111,6 @@ 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),
@@ -230,19 +229,8 @@ void zmq::session_base_t::clean_pipes ()
 
 void zmq::session_base_t::terminated (pipe_t *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;
-    
+    //  Drop the reference to the deallocated pipe.
+    zmq_assert (pipe == pipe_);
     pipe = NULL;
 
     //  If we are waiting for pending messages to be sent, at this point
@@ -254,10 +242,6 @@ 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))
@@ -268,10 +252,6 @@ 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)
@@ -311,7 +291,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 || incomplete_detach > 0) && !is_terminating ()) {
+    if (!pipe && !is_terminating ()) {
         object_t *parents [2] = {this, socket};
         pipe_t *pipes [2] = {NULL, NULL};
         int hwms [2] = {options.rcvhwm, options.sndhwm};
@@ -328,6 +308,9 @@ 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.
@@ -415,17 +398,8 @@ void zmq::session_base_t::detached ()
         return;
     }
 
-    //  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);
@@ -434,6 +408,14 @@ 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;
         
-        //  This flag is set if we are disconnecting, but haven't yet completed
-        int incomplete_detach;
+        //  Socket end of the pipe, for delay attach scenario
+        zmq::pipe_t *outpipe;
 
         //  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_, (struct sockaddr*) &addr, sizeof (addr));
+    rc = connect (*w_, (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.
-    struct sockaddr_in lcladdr;
+    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,11 +970,7 @@ void zmq::socket_base_t::write_activated (pipe_t *pipe_)
 
 void zmq::socket_base_t::hiccuped (pipe_t *pipe_)
 {
-    if (options.delay_attach_on_connect == 1) 
-        pipe_->terminate (false);
-    else 
-        //  Notify derived sockets of the hiccup
-        xhiccuped (pipe_);
+    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,7 +29,6 @@
 #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,7 +31,6 @@
 #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,111 +27,6 @@
 
 #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");
@@ -174,7 +69,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;
@@ -244,19 +139,9 @@ int main (int argc, char *argv [])
 
     rc = zmq_close (to);
     assert (rc == 0);
-    
-    rc = zmq_ctx_destroy(context);
-    assert (rc == 0);
-
-    fprintf (stderr, " Running DELAY_ATTACH_ON_CONNECT with disconnect\n");
 
-    pthread_t serv, work;
-
-    rc = pthread_create (&serv, NULL, server, NULL);
+    rc = zmq_ctx_destroy(context);
     assert (rc == 0);
 
-    rc = pthread_create (&work, NULL, worker, NULL);
-    assert (rc == 0);
-    
-    pthread_exit(NULL);
+    return 0;
 }

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://0mq.is.teh.best:1234");
+    rc = zmq_connect (sock, "tcp://foobar123xyz:1234");
     assert (rc == -1);
     assert (errno == EINVAL);