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

.gitignore

+syntax: glob # for hg-git users
 Makefile
 Makefile.in
 configure

.hgeol

+[patterns]
+** = native

AUTHORS

@@ -59,11 +59,13 @@ Pavel Gushcha <pavimus@gmail.com>
 Pavol Malosek <malosek@fastmq.com>
 Perry Kundert <perry@kundert.ca>
 Peter Bourgon <peter.bourgon@gmail.com>
+Philip Kovacs <phil@philkovacs.com>
 Pieter Hintjens <ph@imatix.com>
 Piotr Trojanek <piotr.trojanek@gmail.com>
 Robert G. Jakabosky <bobby@sharedrealm.com>
 Sebastian Otaegui <feniix@gmail.com>
 Steven McCoy <steven.mccoy@miru.hk>
+Stuart Webster <sw_webster@hotmail.com>
 Tamara Kustarova <kustarova.tamara@gmail.com>
 Taras Shpot <taras.shpot@eleks.com>
 Tero Marttila <terom@fixme.fi>

doc/Makefile.am

@@ -29,11 +29,11 @@ if BUILD_DOC
 SUFFIXES=.html .txt .xml .3 .7
 
 .txt.html:
-	asciidoc -d manpage -b xhtml11 -f asciidoc.conf \
-		-azmq_version=@PACKAGE_VERSION@ $<
+	asciidoc -d manpage -b xhtml11 -f $(srcdir)/asciidoc.conf \
+		-azmq_version=@PACKAGE_VERSION@ -o$@ $<
 .txt.xml:
-	asciidoc -d manpage -b docbook -f asciidoc.conf \
-		-azmq_version=@PACKAGE_VERSION@ $<
+	asciidoc -d manpage -b docbook -f $(srcdir)/asciidoc.conf \
+		-azmq_version=@PACKAGE_VERSION@ -o$@ $<
 .xml.1:
 	xmlto man $<
 .xml.3:

doc/zmq_getsockopt.txt

@@ -331,7 +331,7 @@ ZMQ_IPV4ONLY: Retrieve IPv4-only socket override status
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Retrieve the underlying native socket type.  A value of `1` will use IPv4
-sockets, while the default value of `0` will use IPv6 sockets. An IPv6 socket
+sockets, while the value of `0` will use IPv6 sockets. An IPv6 socket
 lets applications connect to and accept connections from both IPv4 and IPv6
 hosts.
 

doc/zmq_setsockopt.txt

@@ -338,7 +338,7 @@ ZMQ_IPV4ONLY: Use IPv4-only sockets
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Sets the underlying native socket type.  A value of `1` will use IPv4 sockets,
-while the default value of `0` will use IPv6 sockets. An IPv6 socket lets 
+while the value of `0` will use IPv6 sockets. An IPv6 socket lets 
 applications connect to and accept connections from both IPv4 and IPv6 hosts.
 
 [horizontal]

perf/Makefile.am

-INCLUDES = -I$(top_builddir)/include
+INCLUDES = -I$(top_builddir)/include \
+           -I$(top_srcdir)/include
 
 noinst_PROGRAMS = local_lat remote_lat local_thr remote_thr inproc_lat inproc_thr
 

src/signaler.cpp

@@ -238,7 +238,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
     //  two instances of the library don't accidentally create signaler
     //  crossing the process boundary.
     //  We'll use named event object to implement the critical section.
-    HANDLE sync = CreateEvent (NULL, FALSE, FALSE, "zmq-signaler-port-sync");
+    HANDLE sync = CreateEvent (NULL, FALSE, TRUE, "zmq-signaler-port-sync");
     win_assert (sync != NULL);
 
     //  Enter the critical section.

src/stream_engine.cpp

@@ -360,7 +360,7 @@ int zmq::stream_engine_t::read (void *data_, size_t size_)
 
     //  Signalise peer failure.
     if (nbytes == -1 && (errno == ECONNRESET || errno == ECONNREFUSED ||
-          errno == ETIMEDOUT || errno == EHOSTUNREACH))
+          errno == ETIMEDOUT || errno == EHOSTUNREACH || errno == ENOTCONN))
         return -1;
 
     errno_assert (nbytes != -1);

src/tcp_address.cpp

@@ -110,7 +110,7 @@ int zmq::tcp_address_t::resolve_nic_name (const char *nic_, bool ipv4only_)
     return 0;
 }
 
-#elif defined ZMQ_HAVE_AIX || ZMQ_HAVE_HPUX || ZMQ_HAVE_ANDROID
+#elif defined ZMQ_HAVE_AIX || defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_ANDROID
 
 #include <sys/types.h>
 #include <unistd.h>

src/xrep.cpp

@@ -189,14 +189,18 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
         return 0;
     }
 
-    //  Get next message part.
-    pipe_t *pipe;
-    int rc = fq.recvpipe (msg_, flags_, &pipe);
-    if (rc != 0)
-        return -1;
-
-    //  If identity is received, change the key assigned to the pipe.
-    if (unlikely (msg_->flags () & msg_t::identity)) {
+    pipe_t *pipe = NULL;
+    while (true) {
+
+        //  Get next message part.
+        int rc = fq.recvpipe (msg_, flags_, &pipe);
+        if (rc != 0)
+            return -1;
+
+        //  If identity is received, change the key assigned to the pipe.
+        if (likely (!(msg_->flags () & msg_t::identity)))
+            break;
+
         zmq_assert (!more_in);
 
         //  Empty identity means we can preserve the auto-generated identity.
@@ -219,11 +223,6 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
             }
             zmq_assert (it != outpipes.end ());
         }
-
-        //  After processing the identity, try to get the next message.
-        rc = fq.recvpipe (msg_, flags_, &pipe);
-        if (rc != 0)
-            return -1;
     }
 
     //  If we are in the middle of reading a message, just return the next part.
@@ -234,7 +233,7 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
  
     //  We are at the beginning of a new message. Move the message part we
     //  have to the prefetched and return the ID of the peer instead.
-    rc = prefetched_msg.move (*msg_);
+    int rc = prefetched_msg.move (*msg_);
     errno_assert (rc == 0);
     prefetched = true;
     rc = msg_->close ();
@@ -260,9 +259,17 @@ int zmq::xrep_t::rollback (void)
 
 bool zmq::xrep_t::xhas_in ()
 {
+    //  We may already have a message pre-fetched.
     if (prefetched)
         return true;
-    return fq.has_in ();
+
+    //  Try to read the next message to the pre-fetch buffer.
+    int rc = xrep_t::xrecv (&prefetched_msg, ZMQ_DONTWAIT);
+    if (rc != 0 && errno == EAGAIN)
+        return false;
+    zmq_assert (rc == 0);
+    prefetched = true;
+    return true;
 }
 
 bool zmq::xrep_t::xhas_out ()

src/xreq.cpp

@@ -24,7 +24,8 @@
 #include "msg.hpp"
 
 zmq::xreq_t::xreq_t (class ctx_t *parent_, uint32_t tid_) :
-    socket_base_t (parent_, tid_)
+    socket_base_t (parent_, tid_),
+    prefetched (false)
 {
     options.type = ZMQ_XREQ;
 
@@ -36,10 +37,13 @@ zmq::xreq_t::xreq_t (class ctx_t *parent_, uint32_t tid_) :
 
     options.send_identity = true;
     options.recv_identity = true;
+
+    prefetched_msg.init ();
 }
 
 zmq::xreq_t::~xreq_t ()
 {
+    prefetched_msg.close ();
 }
 
 void zmq::xreq_t::xattach_pipe (pipe_t *pipe_)
@@ -56,6 +60,14 @@ int zmq::xreq_t::xsend (msg_t *msg_, int flags_)
 
 int zmq::xreq_t::xrecv (msg_t *msg_, int flags_)
 {
+    //  If there is a prefetched message, return it.
+    if (prefetched) {
+        int rc = msg_->move (prefetched_msg);
+        errno_assert (rc == 0);
+        prefetched = false;
+        return 0;
+    }
+
     //  XREQ socket doesn't use identities. We can safely drop it and 
     while (true) {
         int rc = fq.recv (msg_, flags_);
@@ -69,7 +81,17 @@ int zmq::xreq_t::xrecv (msg_t *msg_, int flags_)
 
 bool zmq::xreq_t::xhas_in ()
 {
-    return fq.has_in ();
+    //  We may already have a message pre-fetched.
+    if (prefetched)
+        return true;
+
+    //  Try to read the next message to the pre-fetch buffer.
+    int rc = xreq_t::xrecv (&prefetched_msg, ZMQ_DONTWAIT);
+    if (rc != 0 && errno == EAGAIN)
+        return false;
+    zmq_assert (rc == 0);
+    prefetched = true;
+    return true;
 }
 
 bool zmq::xreq_t::xhas_out ()

src/xreq.hpp

@@ -62,6 +62,12 @@ namespace zmq
         fq_t fq;
         lb_t lb;
 
+        //  Have we prefetched a message.
+        bool prefetched;
+
+        //  Holds the prefetched message.
+        msg_t prefetched_msg;
+
         xreq_t (const xreq_t&);
         const xreq_t &operator = (const xreq_t&);
     };

tests/Makefile.am

-INCLUDES = -I$(top_builddir)/include
+INCLUDES = -I$(top_builddir)/include \
+           -I$(top_srcdir)/include
+
 LDADD = $(top_builddir)/src/libzmq.la
 
 noinst_PROGRAMS = test_pair_inproc \