From 5ac63140b01fed145fa41f613308e134420920ab Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Thu, 28 Jul 2011 13:19:55 +0200 Subject: Implementations of TCP and IPC transports separated Signed-off-by: Martin Sustrik --- src/Makefile.am | 4 + src/ipc_connecter.cpp | 381 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/ipc_connecter.hpp | 112 +++++++++++++++ src/ipc_listener.cpp | 351 ++++++++++++++++++++++++++++++++++++++++++++++ src/ipc_listener.hpp | 83 +++++++++++ src/session.cpp | 16 ++- src/socket_base.cpp | 16 ++- src/tcp_connecter.cpp | 46 +++--- src/tcp_connecter.hpp | 4 +- src/tcp_engine.cpp | 20 --- src/tcp_listener.cpp | 20 +++ 11 files changed, 1008 insertions(+), 45 deletions(-) create mode 100644 src/ipc_connecter.cpp create mode 100644 src/ipc_connecter.hpp create mode 100644 src/ipc_listener.cpp create mode 100644 src/ipc_listener.hpp diff --git a/src/Makefile.am b/src/Makefile.am index 1da5be9..9f5ea65 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -24,6 +24,8 @@ libzmq_la_SOURCES = \ io_object.hpp \ io_thread.hpp \ ip.hpp \ + ipc_connecter.hpp \ + ipc_listener.hpp \ i_engine.hpp \ i_poll_events.hpp \ kqueue.hpp \ @@ -86,6 +88,8 @@ libzmq_la_SOURCES = \ io_object.cpp \ io_thread.cpp \ ip.cpp \ + ipc_connecter.cpp \ + ipc_listener.cpp \ kqueue.cpp \ lb.cpp \ mailbox.cpp \ diff --git a/src/ipc_connecter.cpp b/src/ipc_connecter.cpp new file mode 100644 index 0000000..942cd49 --- /dev/null +++ b/src/ipc_connecter.cpp @@ -0,0 +1,381 @@ +/* + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include +#include + +#include "ipc_connecter.hpp" +#include "tcp_engine.hpp" +#include "io_thread.hpp" +#include "platform.hpp" +#include "random.hpp" +#include "ip.hpp" +#include "err.hpp" + +#if defined ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#else +#include +#include +#include +#include +#include +#include +#include +#include +#ifdef ZMQ_HAVE_OPENVMS +#include +#endif +#endif + +zmq::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_, + class session_t *session_, const options_t &options_, + const char *address_, bool wait_) : + own_t (io_thread_, options_), + io_object_t (io_thread_), + s (retired_fd), + handle_valid (false), + wait (wait_), + session (session_), + current_reconnect_ivl(options.reconnect_ivl) +{ + memset (&addr, 0, sizeof (addr)); + addr_len = 0; + + // TODO: set_addess should be called separately, so that the error + // can be propagated. + int rc = set_address (address_); + zmq_assert (rc == 0); +} + +zmq::ipc_connecter_t::~ipc_connecter_t () +{ + if (wait) + cancel_timer (reconnect_timer_id); + if (handle_valid) + rm_fd (handle); + + if (s != retired_fd) + close (); +} + +void zmq::ipc_connecter_t::process_plug () +{ + if (wait) + add_reconnect_timer(); + else + start_connecting (); +} + +void zmq::ipc_connecter_t::in_event () +{ + // We are not polling for incomming data, so we are actually called + // because of error here. However, we can get error on out event as well + // on some platforms, so we'll simply handle both events in the same way. + out_event (); +} + +void zmq::ipc_connecter_t::out_event () +{ + fd_t fd = connect (); + rm_fd (handle); + handle_valid = false; + + // Handle the error condition by attempt to reconnect. + if (fd == retired_fd) { + close (); + wait = true; + add_reconnect_timer(); + return; + } + + // Create the engine object for this connection. + tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options); + alloc_assert (engine); + + // Attach the engine to the corresponding session object. + send_attach (session, engine); + + // Shut the connecter down. + terminate (); +} + +void zmq::ipc_connecter_t::timer_event (int id_) +{ + zmq_assert (id_ == reconnect_timer_id); + wait = false; + start_connecting (); +} + +void zmq::ipc_connecter_t::start_connecting () +{ + // Open the connecting socket. + int rc = open (); + + // Connect may succeed in synchronous manner. + if (rc == 0) { + handle = add_fd (s); + handle_valid = true; + out_event (); + return; + } + + // Connection establishment may be dealyed. Poll for its completion. + else if (rc == -1 && errno == EAGAIN) { + handle = add_fd (s); + handle_valid = true; + set_pollout (handle); + return; + } + + // Handle any other error condition by eventual reconnect. + wait = true; + add_reconnect_timer(); +} + +void zmq::ipc_connecter_t::add_reconnect_timer() +{ + add_timer (get_new_reconnect_ivl(), reconnect_timer_id); +} + +int zmq::ipc_connecter_t::get_new_reconnect_ivl () +{ + // The new interval is the current interval + random value. + int this_interval = current_reconnect_ivl + + (generate_random () % options.reconnect_ivl); + + // Only change the current reconnect interval if the maximum reconnect + // interval was set and if it's larger than the reconnect interval. + if (options.reconnect_ivl_max > 0 && + options.reconnect_ivl_max > options.reconnect_ivl) { + + // Calculate the next interval + current_reconnect_ivl = current_reconnect_ivl * 2; + if(current_reconnect_ivl >= options.reconnect_ivl_max) { + current_reconnect_ivl = options.reconnect_ivl_max; + } + } + return this_interval; +} + +#ifdef ZMQ_HAVE_WINDOWS + +int zmq::ipc_connecter_t::set_address (const char *protocol_, const char *addr_) +{ + errno = EPROTONOSUPPORT; + return -1; +} + +int zmq::ipc_connecter_t::open () +{ + zmq_assert (s == retired_fd); + + // Create the socket. + s = socket (addr.ss_family, SOCK_STREAM, IPPROTO_TCP); + if (s == INVALID_SOCKET) { + wsa_error_to_errno (); + return -1; + } + + // Set to non-blocking mode. + unsigned long argp = 1; + int rc = ioctlsocket (s, FIONBIO, &argp); + wsa_assert (rc != SOCKET_ERROR); + + // Connect to the remote peer. + rc = ::connect (s, (sockaddr*) &addr, addr_len); + + // Connect was successfull immediately. + if (rc == 0) + return 0; + + // Asynchronous connect was launched. + if (rc == SOCKET_ERROR && (WSAGetLastError () == WSAEINPROGRESS || + WSAGetLastError () == WSAEWOULDBLOCK)) { + errno = EAGAIN; + return -1; + } + + wsa_error_to_errno (); + return -1; +} + +int zmq::ipc_connecter_t::close () +{ + zmq_assert (s != retired_fd); + int rc = closesocket (s); + wsa_assert (rc != SOCKET_ERROR); + s = retired_fd; + return 0; +} + +zmq::fd_t zmq::ipc_connecter_t::connect () +{ + // Nonblocking connect have finished. Check whether an error occured. + int err = 0; + socklen_t len = sizeof err; + int rc = getsockopt (s, SOL_SOCKET, SO_ERROR, (char*) &err, &len); + zmq_assert (rc == 0); + if (err != 0) { + + // Assert that the error was caused by the networking problems + // rather than 0MQ bug. + if (err == WSAECONNREFUSED || err == WSAETIMEDOUT || + err == WSAECONNABORTED || err == WSAEHOSTUNREACH || + err == WSAENETUNREACH || err == WSAENETDOWN) + return retired_fd; + + wsa_assert_no (err); + } + + // Return the newly connected socket. + fd_t result = s; + s = retired_fd; + return result; +} + +#else + +int zmq::ipc_connecter_t::set_address (const char *addr_) +{ + return resolve_local_path (&addr, &addr_len, addr_); +} + +int zmq::ipc_connecter_t::open () +{ + zmq_assert (s == retired_fd); + struct sockaddr *sa = (struct sockaddr*) &addr; + + if (AF_UNIX != sa->sa_family) { + + // Create the socket. + s = socket (sa->sa_family, SOCK_STREAM, IPPROTO_TCP); + if (s == -1) + return -1; + + // Set to non-blocking mode. +#ifdef ZMQ_HAVE_OPENVMS + int flags = 1; + int rc = ioctl (s, FIONBIO, &flags); + errno_assert (rc != -1); +#else + int flags = fcntl (s, F_GETFL, 0); + if (flags == -1) + flags = 0; + int rc = fcntl (s, F_SETFL, flags | O_NONBLOCK); + errno_assert (rc != -1); +#endif + + // Connect to the remote peer. + rc = ::connect (s, (struct sockaddr*) &addr, addr_len); + + // Connect was successfull immediately. + if (rc == 0) + return 0; + + // Asynchronous connect was launched. + if (rc == -1 && errno == EINPROGRESS) { + errno = EAGAIN; + return -1; + } + + // Error occured. + int err = errno; + close (); + errno = err; + return -1; + } + +#ifndef ZMQ_HAVE_OPENVMS + else { + + // Create the socket. + zmq_assert (AF_UNIX == sa->sa_family); + s = socket (AF_UNIX, SOCK_STREAM, 0); + if (s == -1) + return -1; + + // Set the non-blocking flag. + int flag = fcntl (s, F_GETFL, 0); + if (flag == -1) + flag = 0; + int rc = fcntl (s, F_SETFL, flag | O_NONBLOCK); + errno_assert (rc != -1); + + // Connect to the remote peer. + rc = ::connect (s, (struct sockaddr*) &addr, sizeof (sockaddr_un)); + + // Connect was successfull immediately. + if (rc == 0) + return 0; + + // Error occured. + int err = errno; + close (); + errno = err; + return -1; + } +#endif + + zmq_assert (false); + return -1; +} + +int zmq::ipc_connecter_t::close () +{ + zmq_assert (s != retired_fd); + int rc = ::close (s); + if (rc != 0) + return -1; + s = retired_fd; + return 0; +} + +zmq::fd_t zmq::ipc_connecter_t::connect () +{ + // Following code should handle both Berkeley-derived socket + // implementations and Solaris. + int err = 0; +#if defined ZMQ_HAVE_HPUX + int len = sizeof (err); +#else + socklen_t len = sizeof (err); +#endif + int rc = getsockopt (s, SOL_SOCKET, SO_ERROR, (char*) &err, &len); + if (rc == -1) + err = errno; + if (err != 0) { + + // Assert if the error was caused by 0MQ bug. + // Networking problems are OK. No need to assert. + errno = err; + errno_assert (errno == ECONNREFUSED || errno == ECONNRESET || + errno == ETIMEDOUT || errno == EHOSTUNREACH || + errno == ENETUNREACH || errno == ENETDOWN); + + return retired_fd; + } + + fd_t result = s; + s = retired_fd; + return result; +} + +#endif diff --git a/src/ipc_connecter.hpp b/src/ipc_connecter.hpp new file mode 100644 index 0000000..272e09c --- /dev/null +++ b/src/ipc_connecter.hpp @@ -0,0 +1,112 @@ +/* + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#ifndef __IPC_CONNECTER_HPP_INCLUDED__ +#define __IPC_CONNECTER_HPP_INCLUDED__ + +#include "fd.hpp" +#include "ip.hpp" +#include "own.hpp" +#include "io_object.hpp" +#include "stdint.hpp" + +namespace zmq +{ + + class ipc_connecter_t : public own_t, public io_object_t + { + public: + + // If 'delay' is true connecter first waits for a while, then starts + // connection process. + ipc_connecter_t (class io_thread_t *io_thread_, + class session_t *session_, const options_t &options_, + const char *address_, bool delay_); + ~ipc_connecter_t (); + + private: + + // ID of the timer used to delay the reconnection. + enum {reconnect_timer_id = 1}; + + // Handlers for incoming commands. + void process_plug (); + + // Handlers for I/O events. + void in_event (); + void out_event (); + void timer_event (int id_); + + // Internal function to start the actual connection establishment. + void start_connecting (); + + // Internal function to add a reconnect timer + void add_reconnect_timer(); + + // Internal function to return a reconnect backoff delay. + // Will modify the current_reconnect_ivl used for next call + // Returns the currently used interval + int get_new_reconnect_ivl (); + + // Set address to connect to. + int set_address (const char *addr_); + + // Open TCP connecting socket. Returns -1 in case of error, + // 0 if connect was successfull immediately and 1 if async connect + // was launched. + int open (); + + // Close the connecting socket. + int close (); + + // Get the file descriptor of newly created connection. Returns + // retired_fd if the connection was unsuccessfull. + fd_t connect (); + + // Address to connect to. + sockaddr_storage addr; + socklen_t addr_len; + + // Underlying socket. + fd_t s; + + // Handle corresponding to the listening socket. + handle_t handle; + + // If true file descriptor is registered with the poller and 'handle' + // contains valid value. + bool handle_valid; + + // If true, connecter is waiting a while before trying to connect. + bool wait; + + // Reference to the session we belong to. + class session_t *session; + + // Current reconnect ivl, updated for backoff strategy + int current_reconnect_ivl; + + ipc_connecter_t (const ipc_connecter_t&); + const ipc_connecter_t &operator = (const ipc_connecter_t&); + }; + +} + +#endif diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp new file mode 100644 index 0000000..e11d522 --- /dev/null +++ b/src/ipc_listener.cpp @@ -0,0 +1,351 @@ +/* + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include + +#include + +#include "ipc_listener.hpp" +#include "platform.hpp" +#include "tcp_engine.hpp" +#include "io_thread.hpp" +#include "session.hpp" +#include "config.hpp" +#include "err.hpp" + +#ifdef ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#else +#include +#include +#include +#include +#include +#include +#include +#ifndef ZMQ_HAVE_OPENVMS +#include +#else +#include +#endif +#endif + +zmq::ipc_listener_t::ipc_listener_t (io_thread_t *io_thread_, + socket_base_t *socket_, const options_t &options_) : + own_t (io_thread_, options_), + io_object_t (io_thread_), + has_file (false), + s (retired_fd), + socket (socket_) +{ + memset (&addr, 0, sizeof (addr)); + addr_len = 0; +} + +zmq::ipc_listener_t::~ipc_listener_t () +{ + if (s != retired_fd) + close (); +} + +void zmq::ipc_listener_t::process_plug () +{ + // Start polling for incoming connections. + handle = add_fd (s); + set_pollin (handle); +} + +void zmq::ipc_listener_t::process_term (int linger_) +{ + rm_fd (handle); + own_t::process_term (linger_); +} + +void zmq::ipc_listener_t::in_event () +{ + fd_t fd = accept (); + + // If connection was reset by the peer in the meantime, just ignore it. + // TODO: Handle specific errors like ENFILE/EMFILE etc. + if (fd == retired_fd) + return; + + // Create the engine object for this connection. + tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options); + alloc_assert (engine); + + // Choose I/O thread to run connecter in. Given that we are already + // running in an I/O thread, there must be at least one available. + io_thread_t *io_thread = choose_io_thread (options.affinity); + zmq_assert (io_thread); + + // Create and launch a session object. + session_t *session = new (std::nothrow) + session_t (io_thread, false, socket, options, NULL, NULL); + alloc_assert (session); + session->inc_seqnum (); + launch_child (session); + send_attach (session, engine, false); +} + +#ifdef ZMQ_HAVE_WINDOWS + +int zmq::ipc_listener_t::set_address (const char *protocol_, const char *addr_) +{ + // IPC protocol is not supported on Windows platform. + if (strcmp (protocol_, "tcp") != 0 ) { + errno = EPROTONOSUPPORT; + return -1; + } + + // Convert the interface into sockaddr_in structure. + int rc = resolve_ip_interface (&addr, &addr_len, addr_); + if (rc != 0) + return rc; + + // Create a listening socket. + s = ::socket (addr.ss_family, SOCK_STREAM, IPPROTO_TCP); + if (s == INVALID_SOCKET) { + wsa_error_to_errno (); + return -1; + } + + // Allow reusing of the address. + int flag = 1; + rc = setsockopt (s, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, + (const char*) &flag, sizeof (int)); + wsa_assert (rc != SOCKET_ERROR); + + // Bind the socket to the network interface and port. + rc = bind (s, (struct sockaddr*) &addr, addr_len); + if (rc == SOCKET_ERROR) { + wsa_error_to_errno (); + return -1; + } + + // Listen for incomming connections. + rc = listen (s, options.backlog); + if (rc == SOCKET_ERROR) { + wsa_error_to_errno (); + return -1; + } + + return 0; +} + +int zmq::ipc_listener_t::close () +{ + zmq_assert (s != retired_fd); + int rc = closesocket (s); + wsa_assert (rc != SOCKET_ERROR); + s = retired_fd; + return 0; +} + +zmq::fd_t zmq::ipc_listener_t::accept () +{ + zmq_assert (s != retired_fd); + + // Accept one incoming connection. + fd_t sock = ::accept (s, NULL, NULL); + if (sock == INVALID_SOCKET && + (WSAGetLastError () == WSAEWOULDBLOCK || + WSAGetLastError () == WSAECONNRESET)) + return retired_fd; + + zmq_assert (sock != INVALID_SOCKET); + + // Set to non-blocking mode. + unsigned long argp = 1; + int rc = ioctlsocket (sock, FIONBIO, &argp); + wsa_assert (rc != SOCKET_ERROR); + + return sock; +} + +#else + +int zmq::ipc_listener_t::set_address (const char *protocol_, const char *addr_) +{ + if (strcmp (protocol_, "tcp") == 0 ) { + + // Resolve the sockaddr to bind to. + int rc = resolve_ip_interface (&addr, &addr_len, addr_); + if (rc != 0) + return -1; + + // Create a listening socket. + s = ::socket (addr.ss_family, SOCK_STREAM, IPPROTO_TCP); + if (s == -1) + return -1; + + // Allow reusing of the address. + int flag = 1; + rc = setsockopt (s, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof (int)); + errno_assert (rc == 0); + + // Bind the socket to the network interface and port. + rc = bind (s, (struct sockaddr*) &addr, addr_len); + if (rc != 0) { + int err = errno; + if (close () != 0) + return -1; + errno = err; + return -1; + } + + // Listen for incomming connections. + rc = listen (s, options.backlog); + if (rc != 0) { + int err = errno; + if (close () != 0) + return -1; + errno = err; + return -1; + } + + return 0; + } +#ifndef ZMQ_HAVE_OPENVMS + else if (strcmp (protocol_, "ipc") == 0) { + + // Get rid of the file associated with the UNIX domain socket that + // may have been left behind by the previous run of the application. + ::unlink (addr_); + + // Convert the address into sockaddr_un structure. + int rc = resolve_local_path (&addr, &addr_len, addr_); + if (rc != 0) + return -1; + + // Create a listening socket. + s = ::socket (AF_UNIX, SOCK_STREAM, 0); + if (s == -1) + return -1; + + // Set the non-blocking flag. + int flag = fcntl (s, F_GETFL, 0); + if (flag == -1) + flag = 0; + rc = fcntl (s, F_SETFL, flag | O_NONBLOCK); + errno_assert (rc != -1); + + // Bind the socket to the file path. + rc = bind (s, (struct sockaddr*) &addr, addr_len); + if (rc != 0) { + int err = errno; + if (close () != 0) + return -1; + errno = err; + return -1; + } + has_file = true; + + // Listen for incomming connections. + rc = listen (s, options.backlog); + if (rc != 0) { + int err = errno; + if (close () != 0) + return -1; + errno = err; + return -1; + } + + return 0; + } +#endif + else { + errno = EPROTONOSUPPORT; + return -1; + } +} + +int zmq::ipc_listener_t::close () +{ + zmq_assert (s != retired_fd); + int rc = ::close (s); + if (rc != 0) + return -1; + s = retired_fd; + +#ifndef ZMQ_HAVE_OPENVMS + // If there's an underlying UNIX domain socket, get rid of the file it + // is associated with. + struct sockaddr_un *su = (struct sockaddr_un*) &addr; + if (AF_UNIX == su->sun_family && has_file) { + rc = ::unlink(su->sun_path); + if (rc != 0) + return -1; + } +#endif + + return 0; +} + +zmq::fd_t zmq::ipc_listener_t::accept () +{ + zmq_assert (s != retired_fd); + + // Accept one incoming connection. + fd_t sock = ::accept (s, NULL, NULL); + +#if (defined ZMQ_HAVE_LINUX || defined ZMQ_HAVE_FREEBSD || \ + defined ZMQ_HAVE_OPENBSD || defined ZMQ_HAVE_OSX || \ + defined ZMQ_HAVE_OPENVMS || defined ZMQ_HAVE_NETBSD || \ + defined ZMQ_HAVE_CYGWIN) + if (sock == -1 && + (errno == EAGAIN || errno == EWOULDBLOCK || + errno == EINTR || errno == ECONNABORTED)) + return retired_fd; +#elif (defined ZMQ_HAVE_SOLARIS || defined ZMQ_HAVE_AIX) + if (sock == -1 && + (errno == EWOULDBLOCK || errno == EINTR || + errno == ECONNABORTED || errno == EPROTO)) + return retired_fd; +#elif defined ZMQ_HAVE_HPUX + if (sock == -1 && + (errno == EAGAIN || errno == EWOULDBLOCK || + errno == EINTR || errno == ECONNABORTED || errno == ENOBUFS)) + return retired_fd; +#elif defined ZMQ_HAVE_QNXNTO + if (sock == -1 && + (errno == EWOULDBLOCK || errno == EINTR || errno == ECONNABORTED)) + return retired_fd; +#endif + + errno_assert (sock != -1); + + // Set to non-blocking mode. +#ifdef ZMQ_HAVE_OPENVMS + int flags = 1; + int rc = ioctl (sock, FIONBIO, &flags); + errno_assert (rc != -1); +#else + int flags = fcntl (s, F_GETFL, 0); + if (flags == -1) + flags = 0; + int rc = fcntl (sock, F_SETFL, flags | O_NONBLOCK); + errno_assert (rc != -1); +#endif + + return sock; +} + +#endif diff --git a/src/ipc_listener.hpp b/src/ipc_listener.hpp new file mode 100644 index 0000000..5d0e1d1 --- /dev/null +++ b/src/ipc_listener.hpp @@ -0,0 +1,83 @@ +/* + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#ifndef __ZMQ_IPC_LISTENER_HPP_INCLUDED__ +#define __ZMQ_IPC_LISTENER_HPP_INCLUDED__ + +#include "fd.hpp" +#include "ip.hpp" +#include "own.hpp" +#include "io_object.hpp" +#include "stdint.hpp" + +namespace zmq +{ + + class ipc_listener_t : public own_t, public io_object_t + { + public: + + ipc_listener_t (class io_thread_t *io_thread_, + class socket_base_t *socket_, const options_t &options_); + ~ipc_listener_t (); + + // Set address to listen on. + int set_address (const char* protocol_, const char *addr_); + + private: + + // Handlers for incoming commands. + void process_plug (); + void process_term (int linger_); + + // Handlers for I/O events. + void in_event (); + + // Close the listening socket. + int close (); + + // Accept the new connection. Returns the file descriptor of the + // newly created connection. The function may return retired_fd + // if the connection was dropped while waiting in the listen backlog. + fd_t accept (); + + // Address to listen on. + sockaddr_storage addr; + socklen_t addr_len; + + // True, if the undelying file for UNIX domain socket exists. + bool has_file; + + // Underlying socket. + fd_t s; + + // Handle corresponding to the listening socket. + handle_t handle; + + // Socket the listerner belongs to. + class socket_base_t *socket; + + ipc_listener_t (const ipc_listener_t&); + const ipc_listener_t &operator = (const ipc_listener_t&); + }; + +} + +#endif diff --git a/src/session.cpp b/src/session.cpp index b857724..5db080f 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -25,6 +25,7 @@ #include "pipe.hpp" #include "likely.hpp" #include "tcp_connecter.hpp" +#include "ipc_connecter.hpp" #include "vtcp_connecter.hpp" #include "pgm_sender.hpp" #include "pgm_receiver.hpp" @@ -305,12 +306,17 @@ void zmq::session_t::start_connecting (bool wait_) // Create the connecter object. - // Both TCP and IPC transports are using the same infrastructure. - if (protocol == "tcp" || protocol == "ipc") { - + if (protocol == "tcp") { tcp_connecter_t *connecter = new (std::nothrow) tcp_connecter_t ( - io_thread, this, options, protocol.c_str (), address.c_str (), - wait_); + io_thread, this, options, address.c_str (), wait_); + alloc_assert (connecter); + launch_child (connecter); + return; + } + + if (protocol == "ipc") { + ipc_connecter_t *connecter = new (std::nothrow) ipc_connecter_t ( + io_thread, this, options, address.c_str (), wait_); alloc_assert (connecter); launch_child (connecter); return; diff --git a/src/socket_base.cpp b/src/socket_base.cpp index fb0f5fd..975934f 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -35,6 +35,7 @@ #include "socket_base.hpp" #include "tcp_listener.hpp" +#include "ipc_listener.hpp" #include "vtcp_listener.hpp" #include "tcp_connecter.hpp" #include "io_thread.hpp" @@ -363,7 +364,7 @@ int zmq::socket_base_t::bind (const char *addr_) return -1; } - if (protocol == "tcp" || protocol == "ipc") { + if (protocol == "tcp") { tcp_listener_t *listener = new (std::nothrow) tcp_listener_t ( io_thread, this, options); alloc_assert (listener); @@ -376,6 +377,19 @@ int zmq::socket_base_t::bind (const char *addr_) return 0; } + if (protocol == "ipc") { + ipc_listener_t *listener = new (std::nothrow) ipc_listener_t ( + io_thread, this, options); + alloc_assert (listener); + int rc = listener->set_address (protocol.c_str(), address.c_str ()); + if (rc != 0) { + delete listener; + return -1; + } + launch_child (listener); + return 0; + } + #if defined ZMQ_HAVE_VTCP if (protocol == "vtcp") { vtcp_listener_t *listener = new (std::nothrow) vtcp_listener_t ( diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp index 6deb145..76ec0bc 100644 --- a/src/tcp_connecter.cpp +++ b/src/tcp_connecter.cpp @@ -47,7 +47,7 @@ zmq::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_, class session_t *session_, const options_t &options_, - const char *protocol_, const char *address_, bool wait_) : + const char *address_, bool wait_) : own_t (io_thread_, options_), io_object_t (io_thread_), s (retired_fd), @@ -59,8 +59,10 @@ zmq::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_, memset (&addr, 0, sizeof (addr)); addr_len = 0; - int rc = set_address (protocol_, address_); - zmq_assert (rc == 0); //TODO: take care ENOMEM, EINVAL + // TODO: set_addess should be called separately, so that the error + // can be propagated. + int rc = set_address (address_); + zmq_assert (rc == 0); } zmq::tcp_connecter_t::~tcp_connecter_t () @@ -104,6 +106,26 @@ void zmq::tcp_connecter_t::out_event () return; } + // 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; + int rc = setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay, + sizeof (int)); +#ifdef ZMQ_HAVE_WINDOWS + wsa_assert (rc != SOCKET_ERROR); +#else + errno_assert (rc == 0); +#endif + +#ifdef ZMQ_HAVE_OPENVMS + // Disable delayed acknowledgements as they hurt latency is serious manner. + int nodelack = 1; + rc = setsockopt (fd, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack, + sizeof (int)); + errno_assert (rc != SOCKET_ERROR); +#endif + // Create the engine object for this connection. tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options); alloc_assert (engine); @@ -175,13 +197,9 @@ int zmq::tcp_connecter_t::get_new_reconnect_ivl () #ifdef ZMQ_HAVE_WINDOWS -int zmq::tcp_connecter_t::set_address (const char *protocol_, const char *addr_) +int zmq::tcp_connecter_t::set_address (const char *addr_) { - if (strcmp (protocol_, "tcp") == 0) - return resolve_ip_hostname (&addr, &addr_len, addr_); - - errno = EPROTONOSUPPORT; - return -1; + return resolve_ip_hostname (&addr, &addr_len, addr_); } int zmq::tcp_connecter_t::open () @@ -254,15 +272,9 @@ zmq::fd_t zmq::tcp_connecter_t::connect () #else -int zmq::tcp_connecter_t::set_address (const char *protocol_, const char *addr_) +int zmq::tcp_connecter_t::set_address (const char *addr_) { - if (strcmp (protocol_, "tcp") == 0) - return resolve_ip_hostname (&addr, &addr_len, addr_); - else if (strcmp (protocol_, "ipc") == 0) - return resolve_local_path (&addr, &addr_len, addr_); - - errno = EPROTONOSUPPORT; - return -1; + return resolve_ip_hostname (&addr, &addr_len, addr_); } int zmq::tcp_connecter_t::open () diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp index 41f2c93..6c7aa1b 100644 --- a/src/tcp_connecter.hpp +++ b/src/tcp_connecter.hpp @@ -38,7 +38,7 @@ namespace zmq // connection process. tcp_connecter_t (class io_thread_t *io_thread_, class session_t *session_, const options_t &options_, - const char *protocol_, const char *address_, bool delay_); + const char *address_, bool delay_); ~tcp_connecter_t (); private: @@ -66,7 +66,7 @@ namespace zmq int get_new_reconnect_ivl (); // Set address to connect to. - int set_address (const char *protocol, const char *addr_); + int set_address (const char *addr_); // Open TCP connecting socket. Returns -1 in case of error, // 0 if connect was successfull immediately and 1 if async connect diff --git a/src/tcp_engine.cpp b/src/tcp_engine.cpp index 4cb535a..f940b84 100644 --- a/src/tcp_engine.cpp +++ b/src/tcp_engine.cpp @@ -99,26 +99,6 @@ zmq::tcp_engine_t::tcp_engine_t (fd_t fd_, const options_t &options_) : rc = setsockopt (s, SOL_SOCKET, SO_NOSIGPIPE, &set, sizeof (int)); errno_assert (rc == 0); #endif - - // 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; - rc = setsockopt (s, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay, - sizeof (int)); -#ifdef ZMQ_HAVE_WINDOWS - wsa_assert (rc != SOCKET_ERROR); -#else - errno_assert (rc == 0); -#endif - -#ifdef ZMQ_HAVE_OPENVMS - // Disable delayed acknowledgements as they hurt latency is serious manner. - int nodelack = 1; - rc = setsockopt (s, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack, - sizeof (int)); - errno_assert (rc != SOCKET_ERROR); -#endif } zmq::tcp_engine_t::~tcp_engine_t () diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp index d334768..360be8c 100644 --- a/src/tcp_listener.cpp +++ b/src/tcp_listener.cpp @@ -87,6 +87,26 @@ void zmq::tcp_listener_t::in_event () if (fd == retired_fd) return; + // 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; + int rc = setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, (char*) &nodelay, + sizeof (int)); +#ifdef ZMQ_HAVE_WINDOWS + wsa_assert (rc != SOCKET_ERROR); +#else + errno_assert (rc == 0); +#endif + +#ifdef ZMQ_HAVE_OPENVMS + // Disable delayed acknowledgements as they hurt latency is serious manner. + int nodelack = 1; + rc = setsockopt (fd, IPPROTO_TCP, TCP_NODELACK, (char*) &nodelack, + sizeof (int)); + errno_assert (rc != SOCKET_ERROR); +#endif + // Create the engine object for this connection. tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options); alloc_assert (engine); -- cgit v1.2.3