Imported Debian patch 2.1.9-1debian/2.1.9-1

27 files changed, 753 insertions, 459 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index ce20225..73eaf44 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -58,6 +58,7 @@ libzmq_la_SOURCES = \
     select.hpp \
     semaphore.hpp \
     session.hpp \
+    signaler.hpp \
     socket_base.hpp \
     stdint.hpp \
     sub.hpp \
@@ -118,6 +119,7 @@ libzmq_la_SOURCES = \
     req.cpp \
     select.cpp \
     session.cpp \
+    signaler.cpp \
     socket_base.cpp \
     sub.cpp \
     swap.cpp \
diff --git a/src/Makefile.in b/src/Makefile.in
index 78e3379..f7a9436 100644
--- a/src/Makefile.in
+++ b/src/Makefile.in
@@ -93,14 +93,15 @@ am_libzmq_la_OBJECTS = libzmq_la-clock.lo libzmq_la-command.lo \
 	libzmq_la-poller_base.lo libzmq_la-pull.lo libzmq_la-push.lo \
 	libzmq_la-reaper.lo libzmq_la-pub.lo libzmq_la-rep.lo \
 	libzmq_la-req.lo libzmq_la-select.lo libzmq_la-session.lo \
-	libzmq_la-socket_base.lo libzmq_la-sub.lo libzmq_la-swap.lo \
-	libzmq_la-tcp_connecter.lo libzmq_la-tcp_listener.lo \
-	libzmq_la-tcp_socket.lo libzmq_la-thread.lo \
-	libzmq_la-transient_session.lo libzmq_la-trie.lo \
-	libzmq_la-uuid.lo libzmq_la-xpub.lo libzmq_la-xrep.lo \
-	libzmq_la-xreq.lo libzmq_la-xsub.lo libzmq_la-zmq.lo \
-	libzmq_la-zmq_connecter.lo libzmq_la-zmq_engine.lo \
-	libzmq_la-zmq_init.lo libzmq_la-zmq_listener.lo
+	libzmq_la-signaler.lo libzmq_la-socket_base.lo \
+	libzmq_la-sub.lo libzmq_la-swap.lo libzmq_la-tcp_connecter.lo \
+	libzmq_la-tcp_listener.lo libzmq_la-tcp_socket.lo \
+	libzmq_la-thread.lo libzmq_la-transient_session.lo \
+	libzmq_la-trie.lo libzmq_la-uuid.lo libzmq_la-xpub.lo \
+	libzmq_la-xrep.lo libzmq_la-xreq.lo libzmq_la-xsub.lo \
+	libzmq_la-zmq.lo libzmq_la-zmq_connecter.lo \
+	libzmq_la-zmq_engine.lo libzmq_la-zmq_init.lo \
+	libzmq_la-zmq_listener.lo
 libzmq_la_OBJECTS = $(am_libzmq_la_OBJECTS)
 AM_V_lt = $(am__v_lt_$(V))
 am__v_lt_ = $(am__v_lt_$(AM_DEFAULT_VERBOSITY))
@@ -350,6 +351,7 @@ libzmq_la_SOURCES = \
     select.hpp \
     semaphore.hpp \
     session.hpp \
+    signaler.hpp \
     socket_base.hpp \
     stdint.hpp \
     sub.hpp \
@@ -410,6 +412,7 @@ libzmq_la_SOURCES = \
     req.cpp \
     select.cpp \
     session.cpp \
+    signaler.cpp \
     socket_base.cpp \
     sub.cpp \
     swap.cpp \
@@ -566,6 +569,7 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-req.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-select.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-session.Plo@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-signaler.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-socket_base.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-sub.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/libzmq_la-swap.Plo@am__quote@
@@ -906,6 +910,14 @@ libzmq_la-session.lo: session.cpp
 @AMDEP_TRUE@@am__fastdepCXX_FALSE@	DEPDIR=$(DEPDIR) $(CXXDEPMODE) $(depcomp) @AMDEPBACKSLASH@
 @am__fastdepCXX_FALSE@	$(LIBTOOL) $(AM_V_lt) --tag=CXX $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CXX) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(libzmq_la_CPPFLAGS) $(CPPFLAGS) $(libzmq_la_CXXFLAGS) $(CXXFLAGS) -c -o libzmq_la-session.lo `test -f 'session.cpp' || echo '$(srcdir)/'`session.cpp
 
+libzmq_la-signaler.lo: signaler.cpp
+@am__fastdepCXX_TRUE@	$(AM_V_CXX)$(LIBTOOL) $(AM_V_lt) --tag=CXX $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CXX) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(libzmq_la_CPPFLAGS) $(CPPFLAGS) $(libzmq_la_CXXFLAGS) $(CXXFLAGS) -MT libzmq_la-signaler.lo -MD -MP -MF $(DEPDIR)/libzmq_la-signaler.Tpo -c -o libzmq_la-signaler.lo `test -f 'signaler.cpp' || echo '$(srcdir)/'`signaler.cpp
+@am__fastdepCXX_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/libzmq_la-signaler.Tpo $(DEPDIR)/libzmq_la-signaler.Plo
+@am__fastdepCXX_FALSE@	$(AM_V_CXX) @AM_BACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	source='signaler.cpp' object='libzmq_la-signaler.lo' libtool=yes @AMDEPBACKSLASH@
+@AMDEP_TRUE@@am__fastdepCXX_FALSE@	DEPDIR=$(DEPDIR) $(CXXDEPMODE) $(depcomp) @AMDEPBACKSLASH@
+@am__fastdepCXX_FALSE@	$(LIBTOOL) $(AM_V_lt) --tag=CXX $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CXX) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(libzmq_la_CPPFLAGS) $(CPPFLAGS) $(libzmq_la_CXXFLAGS) $(CXXFLAGS) -c -o libzmq_la-signaler.lo `test -f 'signaler.cpp' || echo '$(srcdir)/'`signaler.cpp
+
 libzmq_la-socket_base.lo: socket_base.cpp
 @am__fastdepCXX_TRUE@	$(AM_V_CXX)$(LIBTOOL) $(AM_V_lt) --tag=CXX $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CXX) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(libzmq_la_CPPFLAGS) $(CPPFLAGS) $(libzmq_la_CXXFLAGS) $(CXXFLAGS) -MT libzmq_la-socket_base.lo -MD -MP -MF $(DEPDIR)/libzmq_la-socket_base.Tpo -c -o libzmq_la-socket_base.lo `test -f 'socket_base.cpp' || echo '$(srcdir)/'`socket_base.cpp
 @am__fastdepCXX_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/libzmq_la-socket_base.Tpo $(DEPDIR)/libzmq_la-socket_base.Plo
diff --git a/src/config.hpp b/src/config.hpp
index f144512..db79284 100644
--- a/src/config.hpp
+++ b/src/config.hpp
@@ -36,6 +36,13 @@ namespace zmq
         //  memory allocation by approximately 99.6%
         message_pipe_granularity = 256,
 
+        //  Commands in pipe per allocation event.
+        command_pipe_granularity = 16,
+
+        //  Size in bytes of the largest message that is still copied around
+        //  rather than being reference-counted.
+        max_vsm_size = 29,
+
         //  Determines how often does socket poll for new commands when it
         //  still has unprocessed messages to handle. Thus, if it is set to 100,
         //  socket will process 100 inbound messages before doing the poll.
diff --git a/src/ctx.cpp b/src/ctx.cpp
index 2758729..9612cd1 100644
--- a/src/ctx.cpp
+++ b/src/ctx.cpp
@@ -18,21 +18,23 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
+#include "platform.hpp"
+#if defined ZMQ_HAVE_WINDOWS
+#include "windows.hpp"
+#endif
+
 #include <new>
 #include <string.h>
 
 #include "ctx.hpp"
 #include "socket_base.hpp"
 #include "io_thread.hpp"
-#include "platform.hpp"
 #include "reaper.hpp"
 #include "err.hpp"
 #include "pipe.hpp"
 
-#if defined ZMQ_HAVE_WINDOWS
-#include "windows.h"
-#else
-#include "unistd.h"
+#if !defined ZMQ_HAVE_WINDOWS
+#include <unistd.h>
 #endif
 
 zmq::ctx_t::ctx_t (uint32_t io_threads_) :
@@ -141,7 +143,7 @@ int zmq::ctx_t::terminate ()
 
     //  Wait till reaper thread closes all the sockets.
     command_t cmd;
-    int rc = term_mailbox.recv (&cmd, true);
+    int rc = term_mailbox.recv (&cmd, -1);
     if (rc == -1 && errno == EINTR)
         return -1;
     zmq_assert (rc == 0);
diff --git a/src/io_thread.cpp b/src/io_thread.cpp
index be52bdd..bb7971d 100644
--- a/src/io_thread.cpp
+++ b/src/io_thread.cpp
@@ -72,7 +72,7 @@ void zmq::io_thread_t::in_event ()
 
         //  Get the next command. If there is none, exit.
         command_t cmd;
-        int rc = mailbox.recv (&cmd, false);
+        int rc = mailbox.recv (&cmd, 0);
         if (rc != 0 && errno == EINTR)
             continue;
         if (rc != 0 && errno == EAGAIN)
diff --git a/src/ip.cpp b/src/ip.cpp
index a63a97d..3ea1f29 100644
--- a/src/ip.cpp
+++ b/src/ip.cpp
@@ -54,7 +54,7 @@ static int resolve_nic_name (in_addr* addr_, char const *interface_)
     size_t ifr_size = sizeof (struct lifreq) * ifn.lifn_count;
     char *ifr = (char*) malloc (ifr_size);
     alloc_assert (ifr);
-    
+
     //  Retrieve interface names.
     lifconf ifc;
     ifc.lifc_family = AF_UNSPEC;
@@ -92,7 +92,7 @@ static int resolve_nic_name (in_addr* addr_, char const *interface_)
     return 0;
 }
 
-#elif defined ZMQ_HAVE_AIX || ZMQ_HAVE_HPUX
+#elif defined ZMQ_HAVE_AIX || ZMQ_HAVE_HPUX || ZMQ_HAVE_ANDROID
 
 #include <sys/types.h>
 #include <unistd.h>
@@ -105,7 +105,7 @@ static int resolve_nic_name (in_addr* addr_, char const *interface_)
     int sd = socket (AF_INET, SOCK_DGRAM, 0);
     zmq_assert (sd != -1);
 
-    struct ifreq ifr; 
+    struct ifreq ifr;
 
     //  Copy interface name for ioctl get.
     strncpy (ifr.ifr_name, interface_, sizeof (ifr.ifr_name));
@@ -123,7 +123,7 @@ static int resolve_nic_name (in_addr* addr_, char const *interface_)
 
     struct sockaddr *sa = (struct sockaddr *) &ifr.ifr_addr;
     *addr_ = ((sockaddr_in*)sa)->sin_addr;
-    return 0;    
+    return 0;
 }
 
 #elif ((defined ZMQ_HAVE_LINUX || defined ZMQ_HAVE_FREEBSD ||\
@@ -140,14 +140,14 @@ static int resolve_nic_name (in_addr* addr_, char const *interface_)
     //  Get the addresses.
     ifaddrs* ifa = NULL;
     int rc = getifaddrs (&ifa);
-    zmq_assert (rc == 0);    
+    zmq_assert (rc == 0);
     zmq_assert (ifa != NULL);
 
     //  Find the corresponding network interface.
     bool found = false;
     for (ifaddrs *ifp = ifa; ifp != NULL ;ifp = ifp->ifa_next)
-        if (ifp->ifa_addr && ifp->ifa_addr->sa_family == AF_INET 
-            && !strcmp (interface_, ifp->ifa_name)) 
+        if (ifp->ifa_addr && ifp->ifa_addr->sa_family == AF_INET
+            && !strcmp (interface_, ifp->ifa_name))
         {
             *addr_ = ((sockaddr_in*) ifp->ifa_addr)->sin_addr;
             found = true;
@@ -294,7 +294,7 @@ int zmq::resolve_ip_hostname (sockaddr_storage *addr_, socklen_t *addr_len_,
     //  Need to choose one to avoid duplicate results from getaddrinfo() - this
     //  doesn't really matter, since it's not included in the addr-output.
     req.ai_socktype = SOCK_STREAM;
-    
+
     //  Avoid named services due to unclear socktype.
     req.ai_flags = AI_NUMERICSERV;
 
@@ -311,9 +311,9 @@ int zmq::resolve_ip_hostname (sockaddr_storage *addr_, socklen_t *addr_len_,
     zmq_assert ((size_t) (res->ai_addrlen) <= sizeof (*addr_));
     memcpy (addr_, res->ai_addr, res->ai_addrlen);
     *addr_len_ = res->ai_addrlen;
- 
+
     freeaddrinfo (res);
-    
+
     return 0;
 }
 
diff --git a/src/mailbox.cpp b/src/mailbox.cpp
index 221396b..a99a9ec 100644
--- a/src/mailbox.cpp
+++ b/src/mailbox.cpp
@@ -21,362 +21,62 @@
 #include "mailbox.hpp"
 #include "platform.hpp"
 #include "err.hpp"
-#include "fd.hpp"
-#include "ip.hpp"
-
-#if defined ZMQ_HAVE_WINDOWS
-#include "windows.hpp"
-#else
-#include <unistd.h>
-#include <fcntl.h>
-#include <limits.h>
-#include <netinet/tcp.h>
-#include <unistd.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#endif
-
-zmq::fd_t zmq::mailbox_t::get_fd ()
-{
-    return r;
-}
-
-#if defined ZMQ_HAVE_WINDOWS
 
 zmq::mailbox_t::mailbox_t ()
 {
-    //  Create the socketpair for signalling.
-    int rc = make_socketpair (&r, &w);
-    errno_assert (rc == 0);
-
-    //  Set the writer to non-blocking mode.
-    unsigned long argp = 1;
-    rc = ioctlsocket (w, FIONBIO, &argp);
-    wsa_assert (rc != SOCKET_ERROR);
-
-    //  Set the reader to non-blocking mode.
-    argp = 1;
-    rc = ioctlsocket (r, FIONBIO, &argp);
-    wsa_assert (rc != SOCKET_ERROR);
+    //  Get the pipe into passive state. That way, if the users starts by
+    //  polling on the associated file descriptor it will get woken up when
+    //  new command is posted.
+    bool ok = cpipe.read (NULL);
+    zmq_assert (!ok);
+    active = false;
 }
 
 zmq::mailbox_t::~mailbox_t ()
 {
-    int rc = closesocket (w);
-    wsa_assert (rc != SOCKET_ERROR);
-
-    rc = closesocket (r);
-    wsa_assert (rc != SOCKET_ERROR);
-}
-
-void zmq::mailbox_t::send (const command_t &cmd_)
-{
-    //  TODO: Implement SNDBUF auto-resizing as for POSIX platforms.
-    //  In the mean time, the following code with assert if the send()
-    //  call would block.
-    int nbytes = ::send (w, (char *)&cmd_, sizeof (command_t), 0);
-    wsa_assert (nbytes != SOCKET_ERROR);
-    zmq_assert (nbytes == sizeof (command_t));
-}
-
-int zmq::mailbox_t::recv (command_t *cmd_, bool block_)
-{
-    //  If required, set the reader to blocking mode.
-    if (block_) {
-        unsigned long argp = 0;
-        int rc = ioctlsocket (r, FIONBIO, &argp);
-        wsa_assert (rc != SOCKET_ERROR);
-    }
-
-    //  Attempt to read an entire command. Returns EAGAIN if non-blocking
-    //  and a command is not available. Save value of errno if we wish to pass
-    //  it to caller.
-    int err = 0;
-    int nbytes = ::recv (r, (char *)cmd_, sizeof (command_t), 0);
-    if (nbytes == -1 && WSAGetLastError () == WSAEWOULDBLOCK)
-        err = EAGAIN;
-
-    //  Re-set the reader to non-blocking mode.
-    if (block_) {
-        unsigned long argp = 1;
-        int rc = ioctlsocket (r, FIONBIO, &argp);
-        wsa_assert (rc != SOCKET_ERROR);
-    }
-
-    //  If the recv failed, return with the saved errno.
-    if (err != 0) {
-        errno = err;
-        return -1;
-    }
-
-    //  Sanity check for success.
-    wsa_assert (nbytes != SOCKET_ERROR);
-
-    //  Check whether we haven't got half of command.
-    zmq_assert (nbytes == sizeof (command_t));
-
-    return 0;
-}
-
-#else
-
-zmq::mailbox_t::mailbox_t ()
-{
-#ifdef PIPE_BUF
-    //  Make sure that command can be written to the socket in atomic fashion.
-    //  If this wasn't guaranteed, commands from different threads would be
-    //  interleaved.
-    zmq_assert (sizeof (command_t) <= PIPE_BUF);
-#endif
-
-    //  Create the socketpair for signaling.
-    int rc = make_socketpair (&r, &w);
-    errno_assert (rc == 0);
-
-    //  Set the writer to non-blocking mode.
-    int flags = fcntl (w, F_GETFL, 0);
-    errno_assert (flags >= 0);
-    rc = fcntl (w, F_SETFL, flags | O_NONBLOCK);
-    errno_assert (rc == 0);
-
-#ifndef MSG_DONTWAIT
-    //  Set the reader to non-blocking mode.
-    flags = fcntl (r, F_GETFL, 0);
-    errno_assert (flags >= 0);
-    rc = fcntl (r, F_SETFL, flags | O_NONBLOCK);
-    errno_assert (rc == 0);
-#endif
+    //  TODO: Retrieve and deallocate commands inside the cpipe.
 }
 
-zmq::mailbox_t::~mailbox_t ()
+zmq::fd_t zmq::mailbox_t::get_fd ()
 {
-    close (w);
-    close (r);
+    return signaler.get_fd ();
 }
 
 void zmq::mailbox_t::send (const command_t &cmd_)
 {
-    //  Attempt to write an entire command without blocking.
-    ssize_t nbytes;
-    do {
-        nbytes = ::send (w, &cmd_, sizeof (command_t), 0);
-    } while (nbytes == -1 && errno == EINTR);
-
-    //  Attempt to increase mailbox SNDBUF if the send failed.
-    if (nbytes == -1 && errno == EAGAIN) {
-        int old_sndbuf, new_sndbuf;
-        socklen_t sndbuf_size = sizeof old_sndbuf;
-
-        //  Retrieve current send buffer size.
-        int rc = getsockopt (w, SOL_SOCKET, SO_SNDBUF, &old_sndbuf,
-            &sndbuf_size);
-        errno_assert (rc == 0);
-        new_sndbuf = old_sndbuf * 2;
-
-        //  Double the new send buffer size.
-        rc = setsockopt (w, SOL_SOCKET, SO_SNDBUF, &new_sndbuf, sndbuf_size);
-        errno_assert (rc == 0);
-
-        //  Verify that the OS actually honored the request.
-        rc = getsockopt (w, SOL_SOCKET, SO_SNDBUF, &new_sndbuf, &sndbuf_size);
-        errno_assert (rc == 0);
-        zmq_assert (new_sndbuf > old_sndbuf);
-
-        //  Retry the sending operation; at this point it must succeed.
-        do {
-            nbytes = ::send (w, &cmd_, sizeof (command_t), 0);
-        } while (nbytes == -1 && errno == EINTR);
-    }
-    errno_assert (nbytes != -1);
-
-    //  This should never happen as we've already checked that command size is
-    //  less than PIPE_BUF.
-    zmq_assert (nbytes == sizeof (command_t));
+    sync.lock ();
+    cpipe.write (cmd_, false);
+    bool ok = cpipe.flush ();
+    sync.unlock ();
+    if (!ok)
+        signaler.send ();
 }
 
-int zmq::mailbox_t::recv (command_t *cmd_, bool block_)
+int zmq::mailbox_t::recv (command_t *cmd_, int timeout_)
 {
-#ifdef MSG_DONTWAIT
-
-    //  Attempt to read an entire command. Returns EAGAIN if non-blocking
-    //  mode is requested and a command is not available.
-    ssize_t nbytes = ::recv (r, cmd_, sizeof (command_t),
-        block_ ? 0 : MSG_DONTWAIT);
-    if (nbytes == -1 && (errno == EAGAIN || errno == EINTR))
-        return -1;
-#else
-
-    //  If required, set the reader to blocking mode.
-    if (block_) {
-        int flags = fcntl (r, F_GETFL, 0);
-        errno_assert (flags >= 0);
-        int rc = fcntl (r, F_SETFL, flags & ~O_NONBLOCK);
-        errno_assert (rc == 0);
-    }
-
-    //  Attempt to read an entire command. Returns EAGAIN if non-blocking
-    //  and a command is not available. Save value of errno if we wish to pass
-    //  it to caller.
-    int err = 0;
-    ssize_t nbytes = ::recv (r, cmd_, sizeof (command_t), 0);
-    if (nbytes == -1 && (errno == EAGAIN || errno == EINTR))  
-        err = errno;
-
-    //  Re-set the reader to non-blocking mode.
-    if (block_) {
-        int flags = fcntl (r, F_GETFL, 0);
-        errno_assert (flags >= 0);
-        int rc = fcntl (r, F_SETFL, flags | O_NONBLOCK);
-        errno_assert (rc == 0);
+    //  Try to get the command straight away.
+    if (active) {
+        bool ok = cpipe.read (cmd_);
+        if (ok)
+            return 0;
+
+        //  If there are no more commands available, switch into passive state.
+        active = false;
+        signaler.recv ();
     }
 
-    //  If the recv failed, return with the saved errno if set.
-    if (err != 0) {
-        errno = err;
+    //  Wait for signal from the command sender.
+    int rc = signaler.wait (timeout_);
+    if (rc != 0 && (errno == EAGAIN || errno == EINTR))
         return -1;
-    }
-
-#endif
-
-    //  Sanity check for success.
-    errno_assert (nbytes != -1);
-
-    //  Check whether we haven't got half of command.
-    zmq_assert (nbytes == sizeof (command_t));
-
-    return 0;
-}
-
-#endif
-
-int zmq::mailbox_t::make_socketpair (fd_t *r_, fd_t *w_)
-{
-#if defined ZMQ_HAVE_WINDOWS
-
-    //  Windows has no 'socketpair' function. CreatePipe is no good as pipe
-    //  handles cannot be polled on. Here we create the socketpair by hand.
-    *w_ = INVALID_SOCKET;
-    *r_ = INVALID_SOCKET;
-
-    //  Create listening socket.
-    SOCKET listener;
-    listener = socket (AF_INET, SOCK_STREAM, 0);
-    wsa_assert (listener != INVALID_SOCKET);
-
-    //  Set SO_REUSEADDR and TCP_NODELAY on listening socket.
-    BOOL so_reuseaddr = 1;
-    int rc = setsockopt (listener, SOL_SOCKET, SO_REUSEADDR,
-        (char *)&so_reuseaddr, sizeof (so_reuseaddr));
-    wsa_assert (rc != SOCKET_ERROR);
-    BOOL tcp_nodelay = 1;
-    rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY,
-        (char *)&tcp_nodelay, sizeof (tcp_nodelay));
-    wsa_assert (rc != SOCKET_ERROR);
-
-    //  Bind listening socket to any free local port.
-    struct sockaddr_in addr;
-    memset (&addr, 0, sizeof (addr));
-    addr.sin_family = AF_INET;
-    addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
-    addr.sin_port = 0;
-    rc = bind (listener, (const struct sockaddr*) &addr, sizeof (addr));
-    wsa_assert (rc != SOCKET_ERROR);
-
-    //  Retrieve local port listener is bound to (into addr).
-    int addrlen = sizeof (addr);
-    rc = getsockname (listener, (struct sockaddr*) &addr, &addrlen);
-    wsa_assert (rc != SOCKET_ERROR);
-
-    //  Listen for incomming connections.
-    rc = listen (listener, 1);
-    wsa_assert (rc != SOCKET_ERROR);
-
-    //  Create the writer socket.
-    *w_ = WSASocket (AF_INET, SOCK_STREAM, 0, NULL, 0,  0);
-    wsa_assert (*w_ != INVALID_SOCKET);
-
-    //  Set TCP_NODELAY on writer socket.
-    rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY,
-        (char *)&tcp_nodelay, sizeof (tcp_nodelay));
-    wsa_assert (rc != SOCKET_ERROR);
 
-    //  Connect writer to the listener.
-    rc = connect (*w_, (sockaddr *) &addr, sizeof (addr));
-    wsa_assert (rc != SOCKET_ERROR);
+    //  We've got the signal. Now we can switch into active state.
+    active = true;
 
-    //  Accept connection from writer.
-    *r_ = accept (listener, NULL, NULL);
-    wsa_assert (*r_ != INVALID_SOCKET);
-
-    //  We don't need the listening socket anymore. Close it.
-    rc = closesocket (listener);
-    wsa_assert (rc != SOCKET_ERROR);
-
-    return 0;
-
-#elif defined ZMQ_HAVE_OPENVMS
-
-    //  Whilst OpenVMS supports socketpair - it maps to AF_INET only.  Further,
-    //  it does not set the socket options TCP_NODELAY and TCP_NODELACK which
-    //  can lead to performance problems.
-    //
-    //  The bug will be fixed in V5.6 ECO4 and beyond.  In the meantime, we'll
-    //  create the socket pair manually.
-    sockaddr_in lcladdr;
-    memset (&lcladdr, 0, sizeof (lcladdr));
-    lcladdr.sin_family = AF_INET;
-    lcladdr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
-    lcladdr.sin_port = 0;
-
-    int listener = socket (AF_INET, SOCK_STREAM, 0);
-    errno_assert (listener != -1);
-
-    int on = 1;
-    int rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on));
-    errno_assert (rc != -1);
-
-    rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on));
-    errno_assert (rc != -1);
-
-    rc = bind(listener, (struct sockaddr*) &lcladdr, sizeof (lcladdr));
-    errno_assert (rc != -1);
-
-    socklen_t lcladdr_len = sizeof (lcladdr);
-
-    rc = getsockname (listener, (struct sockaddr*) &lcladdr, &lcladdr_len);
-    errno_assert (rc != -1);
-
-    rc = listen (listener, 1);
-    errno_assert (rc != -1);
-
-    *w_ = socket (AF_INET, SOCK_STREAM, 0);
-    errno_assert (*w_ != -1);
-
-    rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on));
-    errno_assert (rc != -1);
-
-    rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on));
-    errno_assert (rc != -1);
-
-    rc = connect (*w_, (struct sockaddr*) &lcladdr, sizeof (lcladdr));
-    errno_assert (rc != -1);
-
-    *r_ = accept (listener, NULL, NULL);
-    errno_assert (*r_ != -1);
-
-    close (listener);
-
-    return 0;
-
-#else // All other implementations support socketpair()
-
-    int sv [2];
-    int rc = socketpair (AF_UNIX, SOCK_STREAM, 0, sv);
+    //  Get a command.
     errno_assert (rc == 0);
-    *w_ = sv [0];
-    *r_ = sv [1];
+    bool ok = cpipe.read (cmd_);
+    zmq_assert (ok);
     return 0;
-
-#endif
 }
 
diff --git a/src/mailbox.hpp b/src/mailbox.hpp
index 96bf4eb..0675b99 100644
--- a/src/mailbox.hpp
+++ b/src/mailbox.hpp
@@ -24,10 +24,12 @@
 #include <stddef.h>
 
 #include "platform.hpp"
+#include "signaler.hpp"
 #include "fd.hpp"
-#include "stdint.hpp"
 #include "config.hpp"
 #include "command.hpp"
+#include "ypipe.hpp"
+#include "mutex.hpp"
 
 namespace zmq
 {
@@ -41,16 +43,26 @@ namespace zmq
 
         fd_t get_fd ();
         void send (const command_t &cmd_);
-        int recv (command_t *cmd_, bool block_);
+        int recv (command_t *cmd_, int timeout_);
         
     private:
 
-        //  Write & read end of the socketpair.
-        fd_t w;
-        fd_t r;
+        //  The pipe to store actual commands.
+        typedef ypipe_t <command_t, command_pipe_granularity> cpipe_t;
+        cpipe_t cpipe;
 
-        //  Platform-dependent function to create a socketpair.
-        static int make_socketpair (fd_t *r_, fd_t *w_);
+        //  Signaler to pass signals from writer thread to reader thread.
+        signaler_t signaler;
+
+        //  There's only one thread receiving from the mailbox, but there
+        //  is arbitrary number of threads sending. Given that ypipe requires
+        //  synchronised access on both of its endpoints, we have to synchronise
+        //  the sending side.
+        mutex_t sync;
+
+        //  True if the underlying pipe is active, ie. when we are allowed to
+        //  read commands from it.
+        bool active;
 
         //  Disable copying of mailbox_t object.
         mailbox_t (const mailbox_t&);
diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp
index 4fadadc..7e7da96 100644
--- a/src/pgm_receiver.cpp
+++ b/src/pgm_receiver.cpp
@@ -60,8 +60,8 @@ int zmq::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
 void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_, i_inout *inout_)
 {
     //  Retrieve PGM fds and start polling.
-    int socket_fd;
-    int waiting_pipe_fd;
+    fd_t socket_fd = retired_fd;
+    fd_t waiting_pipe_fd = retired_fd;
     pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
     socket_handle = add_fd (socket_fd);
     pipe_handle = add_fd (waiting_pipe_fd);
diff --git a/src/pgm_sender.cpp b/src/pgm_sender.cpp
index 4d76433..7a0c951 100644
--- a/src/pgm_sender.cpp
+++ b/src/pgm_sender.cpp
@@ -64,10 +64,10 @@ int zmq::pgm_sender_t::init (bool udp_encapsulation_, const char *network_)
 void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, i_inout *inout_)
 {
     //  Alocate 2 fds for PGM socket.
-    int downlink_socket_fd = 0;
-    int uplink_socket_fd = 0;
-    int rdata_notify_fd = 0;
-    int pending_notify_fd = 0;
+    fd_t downlink_socket_fd = retired_fd;
+    fd_t uplink_socket_fd = retired_fd;
+    fd_t rdata_notify_fd = retired_fd;
+    fd_t pending_notify_fd = retired_fd;
 
     encoder.set_inout (inout_);
 
diff --git a/src/pgm_socket.cpp b/src/pgm_socket.cpp
index 5a82907..29ff3e6 100644
--- a/src/pgm_socket.cpp
+++ b/src/pgm_socket.cpp
@@ -394,8 +394,8 @@ zmq::pgm_socket_t::~pgm_socket_t ()
 
 //  Get receiver fds. receive_fd_ is signaled for incoming packets,
 //  waiting_pipe_fd_ is signaled for state driven events and data.
-void zmq::pgm_socket_t::get_receiver_fds (int *receive_fd_, 
-    int *waiting_pipe_fd_)
+void zmq::pgm_socket_t::get_receiver_fds (fd_t *receive_fd_, 
+    fd_t *waiting_pipe_fd_)
 {
     socklen_t socklen;
     bool rc;
@@ -421,8 +421,8 @@ void zmq::pgm_socket_t::get_receiver_fds (int *receive_fd_,
 //  receive_fd_ is for incoming back-channel protocol packets.
 //  rdata_notify_fd_ is raised for waiting repair transmissions.
 //  pending_notify_fd_ is for state driven events.
-void zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_, 
-    int *rdata_notify_fd_, int *pending_notify_fd_)
+void zmq::pgm_socket_t::get_sender_fds (fd_t *send_fd_, fd_t *receive_fd_, 
+    fd_t *rdata_notify_fd_, fd_t *pending_notify_fd_)
 {
     socklen_t socklen;
     bool rc;
diff --git a/src/pgm_socket.hpp b/src/pgm_socket.hpp
index 9699aed..5ba8387 100644
--- a/src/pgm_socket.hpp
+++ b/src/pgm_socket.hpp
@@ -36,6 +36,7 @@
 #include <pgm/in.h>
 #endif
 
+#include "fd.hpp"
 #include "options.hpp"
 
 namespace zmq
@@ -56,12 +57,12 @@ namespace zmq
         int init (bool udp_encapsulation_, const char *network_);
         
         //   Get receiver fds and store them into user allocated memory.
-        void get_receiver_fds (int *receive_fd_, int *waiting_pipe_fd_);
+        void get_receiver_fds (fd_t *receive_fd_, fd_t *waiting_pipe_fd_);
 
         //   Get sender and receiver fds and store it to user allocated 
         //   memory. Receive fd is used to process NAKs from peers.
-        void get_sender_fds (int *send_fd_, int *receive_fd_,
-            int *rdata_notify_fd_, int *pending_notify_fd_);
+        void get_sender_fds (fd_t *send_fd_, fd_t *receive_fd_,
+            fd_t *rdata_notify_fd_, fd_t *pending_notify