From 75af6aed482ab16997c1388fe801f74d11ec12a4 Mon Sep 17 00:00:00 2001 From: Martin Lucina Date: Mon, 23 Jan 2012 08:54:14 +0100 Subject: Imported Upstream version 2.1.9 --- src/mailbox.cpp | 366 +++++--------------------------------------------------- 1 file changed, 33 insertions(+), 333 deletions(-) (limited to 'src/mailbox.cpp') 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 -#include -#include -#include -#include -#include -#include -#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 } -- cgit v1.2.3