diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/Makefile.am | 2 | ||||
| -rw-r--r-- | src/Makefile.in | 28 | ||||
| -rw-r--r-- | src/config.hpp | 7 | ||||
| -rw-r--r-- | src/ctx.cpp | 14 | ||||
| -rw-r--r-- | src/io_thread.cpp | 2 | ||||
| -rw-r--r-- | src/ip.cpp | 20 | ||||
| -rw-r--r-- | src/mailbox.cpp | 366 | ||||
| -rw-r--r-- | src/mailbox.hpp | 26 | ||||
| -rw-r--r-- | src/pgm_receiver.cpp | 4 | ||||
| -rw-r--r-- | src/pgm_sender.cpp | 8 | ||||
| -rw-r--r-- | src/pgm_socket.cpp | 8 | ||||
| -rw-r--r-- | src/pgm_socket.hpp | 7 | ||||
| -rw-r--r-- | src/platform.hpp.in | 3 | ||||
| -rw-r--r-- | src/reaper.cpp | 2 | ||||
| -rw-r--r-- | src/rep.cpp | 2 | ||||
| -rw-r--r-- | src/select.cpp | 9 | ||||
| -rw-r--r-- | src/signaler.cpp | 344 | ||||
| -rw-r--r-- | src/signaler.hpp | 63 | ||||
| -rw-r--r-- | src/socket_base.cpp | 79 | ||||
| -rw-r--r-- | src/socket_base.hpp | 6 | ||||
| -rw-r--r-- | src/swap.cpp | 8 | ||||
| -rw-r--r-- | src/tcp_connecter.cpp | 3 | ||||
| -rw-r--r-- | src/tcp_socket.cpp | 5 | ||||
| -rw-r--r-- | src/thread.cpp | 16 | ||||
| -rw-r--r-- | src/windows.hpp | 152 | ||||
| -rw-r--r-- | src/xrep.cpp | 4 | ||||
| -rw-r--r-- | src/zmq.cpp | 24 |
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) @@ -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_fd_); // Send data as one APDU, transmit window owned memory. size_t send (unsigned char *data_, size_t data_len_); diff --git a/src/platform.hpp.in b/src/platform.hpp.in index 9ac777f..c89739d 100644 --- a/ |