tcp.cpp

Go to the documentation of this file.

/* SPDX-License-Identifier: MPL-2.0 */
 
#include "precompiled.hpp"
#include "macros.hpp"
#include "ip.hpp"
#include "tcp.hpp"
#include "err.hpp"
#include "options.hpp"
 
#if !defined ZMQ_HAVE_WINDOWS
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <unistd.h>
#ifdef ZMQ_HAVE_VXWORKS
#include <sockLib.h>
#endif
#endif
 
#if defined ZMQ_HAVE_OPENVMS
#include <ioctl.h>
#endif
 
#ifdef __APPLE__
#include <TargetConditionals.h>
#endif
 
int 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;
    const int rc =
      setsockopt (s_, IPPROTO_TCP, TCP_NODELAY,
                  reinterpret_cast<char *> (&nodelay), sizeof (int));
    assert_success_or_recoverable (s_, rc);
    if (rc != 0)
        return rc;
 
#ifdef ZMQ_HAVE_OPENVMS
    //  Disable delayed acknowledgements as they hurt latency significantly.
    int nodelack = 1;
    rc = setsockopt (s_, IPPROTO_TCP, TCP_NODELACK, (char *) &nodelack,
                     sizeof (int));
    assert_success_or_recoverable (s_, rc);
#endif
    return rc;
}
 
int zmq::set_tcp_send_buffer (fd_t sockfd_, int bufsize_)
{
    const int rc =
      setsockopt (sockfd_, SOL_SOCKET, SO_SNDBUF,
                  reinterpret_cast<char *> (&bufsize_), sizeof bufsize_);
    assert_success_or_recoverable (sockfd_, rc);
    return rc;
}
 
int zmq::set_tcp_receive_buffer (fd_t sockfd_, int bufsize_)
{
    const int rc =
      setsockopt (sockfd_, SOL_SOCKET, SO_RCVBUF,
                  reinterpret_cast<char *> (&bufsize_), sizeof bufsize_);
    assert_success_or_recoverable (sockfd_, rc);
    return rc;
}
 
int zmq::tune_tcp_keepalives (fd_t s_,
                              int keepalive_,
                              int keepalive_cnt_,
                              int keepalive_idle_,
                              int keepalive_intvl_)
{
    // These options are used only under certain #ifdefs below.
    LIBZMQ_UNUSED (keepalive_);
    LIBZMQ_UNUSED (keepalive_cnt_);
    LIBZMQ_UNUSED (keepalive_idle_);
    LIBZMQ_UNUSED (keepalive_intvl_);
 
    // If none of the #ifdefs apply, then s_ is unused.
    LIBZMQ_UNUSED (s_);
 
    //  Tuning TCP keep-alives if platform allows it
    //  All values = -1 means skip and leave it for OS
#ifdef ZMQ_HAVE_WINDOWS
    if (keepalive_ != -1) {
        tcp_keepalive keepalive_opts;
        keepalive_opts.onoff = keepalive_;
        keepalive_opts.keepalivetime =
          keepalive_idle_ != -1 ? keepalive_idle_ * 1000 : 7200000;
        keepalive_opts.keepaliveinterval =
          keepalive_intvl_ != -1 ? keepalive_intvl_ * 1000 : 1000;
        DWORD num_bytes_returned;
        const int rc = WSAIoctl (s_, SIO_KEEPALIVE_VALS, &keepalive_opts,
                                 sizeof (keepalive_opts), NULL, 0,
                                 &num_bytes_returned, NULL, NULL);
        assert_success_or_recoverable (s_, rc);
        if (rc == SOCKET_ERROR)
            return rc;
    }
#else
#ifdef ZMQ_HAVE_SO_KEEPALIVE
    if (keepalive_ != -1) {
        int rc =
          setsockopt (s_, SOL_SOCKET, SO_KEEPALIVE,
                      reinterpret_cast<char *> (&keepalive_), sizeof (int));
        assert_success_or_recoverable (s_, rc);
        if (rc != 0)
            return rc;
 
#ifdef ZMQ_HAVE_TCP_KEEPCNT
        if (keepalive_cnt_ != -1) {
            int rc = setsockopt (s_, IPPROTO_TCP, TCP_KEEPCNT, &keepalive_cnt_,
                                 sizeof (int));
            assert_success_or_recoverable (s_, rc);
            if (rc != 0)
                return rc;
        }
#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));
            assert_success_or_recoverable (s_, rc);
            if (rc != 0)
                return rc;
        }
#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));
            assert_success_or_recoverable (s_, rc);
            if (rc != 0)
                return rc;
        }
#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));
            assert_success_or_recoverable (s_, rc);
            if (rc != 0)
                return rc;
        }
#endif // ZMQ_HAVE_TCP_KEEPINTVL
    }
#endif // ZMQ_HAVE_SO_KEEPALIVE
#endif // ZMQ_HAVE_WINDOWS
 
    return 0;
}
 
int zmq::tune_tcp_maxrt (fd_t sockfd_, int timeout_)
{
    if (timeout_ <= 0)
        return 0;
 
    LIBZMQ_UNUSED (sockfd_);
 
#if defined(ZMQ_HAVE_WINDOWS) && defined(TCP_MAXRT)
    // msdn says it's supported in >= Vista, >= Windows Server 2003
    timeout_ /= 1000; // in seconds
    const int rc =
      setsockopt (sockfd_, IPPROTO_TCP, TCP_MAXRT,
                  reinterpret_cast<char *> (&timeout_), sizeof (timeout_));
    assert_success_or_recoverable (sockfd_, rc);
    return rc;
// FIXME: should be ZMQ_HAVE_TCP_USER_TIMEOUT
#elif defined(TCP_USER_TIMEOUT)
    int rc = setsockopt (sockfd_, IPPROTO_TCP, TCP_USER_TIMEOUT, &timeout_,
                         sizeof (timeout_));
    assert_success_or_recoverable (sockfd_, rc);
    return rc;
#else
    return 0;
#endif
}
 
int zmq::tcp_write (fd_t s_, const void *data_, size_t size_)
{
#ifdef ZMQ_HAVE_WINDOWS
 
    const int nbytes = send (s_, (char *) data_, static_cast<int> (size_), 0);
 
    //  If not a single byte can be written to the socket in non-blocking mode
    //  we'll get an error (this may happen during the speculative write).
    const int last_error = WSAGetLastError ();
    if (nbytes == SOCKET_ERROR && last_error == WSAEWOULDBLOCK)
        return 0;
 
    //  Signalise peer failure.
    if (nbytes == SOCKET_ERROR
        && (last_error == WSAENETDOWN || last_error == WSAENETRESET
            || last_error == WSAEHOSTUNREACH || last_error == WSAECONNABORTED
            || last_error == WSAETIMEDOUT || last_error == WSAECONNRESET))
        return -1;
 
    //  Circumvent a Windows bug:
    //  See https://support.microsoft.com/en-us/kb/201213
    //  See https://zeromq.jira.com/browse/LIBZMQ-195
    if (nbytes == SOCKET_ERROR && last_error == WSAENOBUFS)
        return 0;
 
    wsa_assert (nbytes != SOCKET_ERROR);
    return nbytes;
 
#else
    ssize_t nbytes = send (s_, static_cast<const char *> (data_), size_, 0);
 
    //  Several errors are OK. When speculative write is being done we may not
    //  be able to write a single byte from the socket. Also, SIGSTOP issued
    //  by a debugging tool can result in EINTR error.
    if (nbytes == -1
        && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
        return 0;
 
    //  Signalise peer failure.
    if (nbytes == -1) {
#if !defined(TARGET_OS_IPHONE) || !TARGET_OS_IPHONE
        errno_assert (errno != EACCES && errno != EBADF && errno != EDESTADDRREQ
                      && errno != EFAULT && errno != EISCONN
                      && errno != EMSGSIZE && errno != ENOMEM
                      && errno != ENOTSOCK && errno != EOPNOTSUPP);
#else
        errno_assert (errno != EACCES && errno != EDESTADDRREQ
                      && errno != EFAULT && errno != EISCONN
                      && errno != EMSGSIZE && errno != ENOMEM
                      && errno != ENOTSOCK && errno != EOPNOTSUPP);
#endif
        return -1;
    }
 
    return static_cast<int> (nbytes);
 
#endif
}
 
int zmq::tcp_read (fd_t s_, void *data_, size_t size_)
{
#ifdef ZMQ_HAVE_WINDOWS
 
    const int rc =
      recv (s_, static_cast<char *> (data_), static_cast<int> (size_), 0);
 
    //  If not a single byte can be read from the socket in non-blocking mode
    //  we'll get an error (this may happen during the speculative read).
    if (rc == SOCKET_ERROR) {
        const int last_error = WSAGetLastError ();
        if (last_error == WSAEWOULDBLOCK) {
            errno = EAGAIN;
        } else {
            wsa_assert (
              last_error == WSAENETDOWN || last_error == WSAENETRESET
              || last_error == WSAECONNABORTED || last_error == WSAETIMEDOUT
              || last_error == WSAECONNRESET || last_error == WSAECONNREFUSED
              || last_error == WSAENOTCONN || last_error == WSAENOBUFS);
            errno = wsa_error_to_errno (last_error);
        }
    }
 
    return rc == SOCKET_ERROR ? -1 : rc;
 
#else
 
    const ssize_t rc = recv (s_, static_cast<char *> (data_), size_, 0);
 
    //  Several errors are OK. When speculative read is being done we may not
    //  be able to read a single byte from the socket. Also, SIGSTOP issued
    //  by a debugging tool can result in EINTR error.
    if (rc == -1) {
#if !defined(TARGET_OS_IPHONE) || !TARGET_OS_IPHONE
        errno_assert (errno != EBADF && errno != EFAULT && errno != ENOMEM
                      && errno != ENOTSOCK);
#else
        errno_assert (errno != EFAULT && errno != ENOMEM && errno != ENOTSOCK);
#endif
        if (errno == EWOULDBLOCK || errno == EINTR)
            errno = EAGAIN;
    }
 
    return static_cast<int> (rc);
 
#endif
}
 
void zmq::tcp_tune_loopback_fast_path (const fd_t socket_)
{
#if defined ZMQ_HAVE_WINDOWS && defined SIO_LOOPBACK_FAST_PATH
    int sio_loopback_fastpath = 1;
    DWORD number_of_bytes_returned = 0;
 
    const int rc = WSAIoctl (
      socket_, SIO_LOOPBACK_FAST_PATH, &sio_loopback_fastpath,
      sizeof sio_loopback_fastpath, NULL, 0, &number_of_bytes_returned, 0, 0);
 
    if (SOCKET_ERROR == rc) {
        const DWORD last_error = ::WSAGetLastError ();
 
        if (WSAEOPNOTSUPP == last_error) {
            // This system is not Windows 8 or Server 2012, and the call is not supported.
        } else {
            wsa_assert (false);
        }
    }
#else
    LIBZMQ_UNUSED (socket_);
#endif
}
 
void zmq::tune_tcp_busy_poll (fd_t socket_, int busy_poll_)
{
#if defined(ZMQ_HAVE_BUSY_POLL)
    if (busy_poll_ > 0) {
        const int rc =
          setsockopt (socket_, SOL_SOCKET, SO_BUSY_POLL,
                      reinterpret_cast<char *> (&busy_poll_), sizeof (int));
        assert_success_or_recoverable (socket_, rc);
    }
#else
    LIBZMQ_UNUSED (socket_);
    LIBZMQ_UNUSED (busy_poll_);
#endif
}
 
zmq::fd_t zmq::tcp_open_socket (const char *address_,
                                const zmq::options_t &options_,
                                bool local_,
                                bool fallback_to_ipv4_,
                                zmq::tcp_address_t *out_tcp_addr_)
{
    //  Convert the textual address into address structure.
    int rc = out_tcp_addr_->resolve (address_, local_, options_.ipv6);
    if (rc != 0)
        return retired_fd;
 
    //  Create the socket.
    fd_t s = open_socket (out_tcp_addr_->family (), SOCK_STREAM, IPPROTO_TCP);
 
    //  IPv6 address family not supported, try automatic downgrade to IPv4.
    if (s == retired_fd && fallback_to_ipv4_
        && out_tcp_addr_->family () == AF_INET6 && errno == EAFNOSUPPORT
        && options_.ipv6) {
        rc = out_tcp_addr_->resolve (address_, local_, false);
        if (rc != 0) {
            return retired_fd;
        }
        s = open_socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
    }
 
    if (s == retired_fd) {
        return retired_fd;
    }
 
    //  On some systems, IPv4 mapping in IPv6 sockets is disabled by default.
    //  Switch it on in such cases.
    if (out_tcp_addr_->family () == AF_INET6)
        enable_ipv4_mapping (s);
 
    // Set the IP Type-Of-Service priority for this socket
    if (options_.tos != 0)
        set_ip_type_of_service (s, options_.tos);
 
    // Set the protocol-defined priority for this socket
    if (options_.priority != 0)
        set_socket_priority (s, options_.priority);
 
    // Set the socket to loopback fastpath if configured.
    if (options_.loopback_fastpath)
        tcp_tune_loopback_fast_path (s);
 
    // Bind the socket to a device if applicable
    if (!options_.bound_device.empty ())
        if (bind_to_device (s, options_.bound_device) == -1)
            goto setsockopt_error;
 
    //  Set the socket buffer limits for the underlying socket.
    if (options_.sndbuf >= 0)
        set_tcp_send_buffer (s, options_.sndbuf);
    if (options_.rcvbuf >= 0)
        set_tcp_receive_buffer (s, options_.rcvbuf);
 
    //  This option removes several delays caused by scheduling, interrupts and context switching.
    if (options_.busy_poll)
        tune_tcp_busy_poll (s, options_.busy_poll);
    return s;
 
setsockopt_error:
#ifdef ZMQ_HAVE_WINDOWS
    rc = closesocket (s);
    wsa_assert (rc != SOCKET_ERROR);
#else
    rc = ::close (s);
    errno_assert (rc == 0);
#endif
    return retired_fd;
}