diff options
| -rw-r--r-- | src/Makefile.am | 2 | ||||
| -rw-r--r-- | src/session.cpp | 13 | ||||
| -rw-r--r-- | src/tcp_connecter.hpp | 3 | ||||
| -rw-r--r-- | src/vtcp_connecter.cpp | 249 | ||||
| -rw-r--r-- | src/vtcp_connecter.hpp | 121 | ||||
| -rw-r--r-- | src/vtcp_listener.cpp | 13 |
6 files changed, 394 insertions, 7 deletions
diff --git a/src/Makefile.am b/src/Makefile.am index 73ba62e..1da5be9 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -64,6 +64,7 @@ libzmq_la_SOURCES = \ tcp_listener.hpp \ thread.hpp \ trie.hpp \ + vtcp_connecter.hpp \ vtcp_listener.hpp \ windows.hpp \ wire.hpp \ @@ -118,6 +119,7 @@ libzmq_la_SOURCES = \ tcp_listener.cpp \ thread.cpp \ trie.cpp \ + vtcp_connecter.cpp \ vtcp_listener.cpp \ xpub.cpp \ xrep.cpp \ diff --git a/src/session.cpp b/src/session.cpp index d14b58c..b857724 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -25,6 +25,7 @@ #include "pipe.hpp" #include "likely.hpp" #include "tcp_connecter.hpp" +#include "vtcp_connecter.hpp" #include "pgm_sender.hpp" #include "pgm_receiver.hpp" @@ -315,6 +316,18 @@ void zmq::session_t::start_connecting (bool wait_) return; } +#if defined ZMQ_HAVE_VTCP + if (protocol == "vtcp") { + + vtcp_connecter_t *connecter = new (std::nothrow) vtcp_connecter_t ( + io_thread, this, options, address.c_str (), + wait_); + alloc_assert (connecter); + launch_child (connecter); + return; + } +#endif + #if defined ZMQ_HAVE_OPENPGM // Both PGM and EPGM transports are using the same infrastructure. diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp index 524e9cc..41f2c93 100644 --- a/src/tcp_connecter.hpp +++ b/src/tcp_connecter.hpp @@ -30,9 +30,6 @@ namespace zmq { - // A base class for different connecters. It handles auto-reconnection - // on behalf of the derived class. - class tcp_connecter_t : public own_t, public io_object_t { public: diff --git a/src/vtcp_connecter.cpp b/src/vtcp_connecter.cpp new file mode 100644 index 0000000..cda469b --- /dev/null +++ b/src/vtcp_connecter.cpp @@ -0,0 +1,249 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ + +#include "vtcp_connecter.hpp" + +#if defined ZMQ_HAVE_VTCP + +#include <new> +#include <string> + +#include "tcp_engine.hpp" +#include "io_thread.hpp" +#include "platform.hpp" +#include "ip.hpp" +#include "err.hpp" + +#if defined ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#else +#include <unistd.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <arpa/inet.h> +#include <netinet/tcp.h> +#include <netinet/in.h> +#include <netdb.h> +#include <fcntl.h> +#ifdef ZMQ_HAVE_OPENVMS +#include <ioctl.h> +#endif +#endif + +zmq::vtcp_connecter_t::vtcp_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; + subport = 0; + + int rc = set_address (address_); + zmq_assert (rc == 0); +} + +zmq::vtcp_connecter_t::~vtcp_connecter_t () +{ + if (wait) + cancel_timer (reconnect_timer_id); + if (handle_valid) + rm_fd (handle); + + if (s != retired_fd) + close (); +} + +int zmq::vtcp_connecter_t::set_address (const char *addr_) +{ + const char *delimiter = strrchr (addr_, '.'); + if (!delimiter) { + errno = EINVAL; + return -1; + } + std::string addr_str (addr_, delimiter - addr_); + std::string subport_str (delimiter + 1); + subport = (vtcp_subport_t) atoi (subport_str.c_str ()); + int rc = resolve_ip_hostname (&addr, &addr_len, addr_str.c_str ()); + if (rc != 0) + return -1; + + return 0; +} + +void zmq::vtcp_connecter_t::process_plug () +{ + if (wait) + add_reconnect_timer(); + else + start_connecting (); +} + +void zmq::vtcp_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::vtcp_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::vtcp_connecter_t::timer_event (int id_) +{ + zmq_assert (id_ == reconnect_timer_id); + wait = false; + start_connecting (); +} + +void zmq::vtcp_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::vtcp_connecter_t::add_reconnect_timer() +{ + add_timer (get_new_reconnect_ivl(), reconnect_timer_id); +} + +int zmq::vtcp_connecter_t::get_new_reconnect_ivl () +{ +#if defined ZMQ_HAVE_WINDOWS + int pid = (int) GetCurrentProcessId (); +#else + int pid = (int) getpid (); +#endif + + // The new interval is the current interval + random value. + int this_interval = current_reconnect_ivl + + ((pid * 13) % 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; +} + +int zmq::vtcp_connecter_t::open () +{ + zmq_assert (s == retired_fd); + + uint16_t port = ntohs (((sockaddr_in*) &addr)->sin_port); + s = vtcp_connect (*(in_addr_t*) &addr, port); + + // Connect was successfull immediately. + if (s != retired_fd) + return 0; + + // Asynchronous connect was launched. + if (errno == EINPROGRESS) { + errno = EAGAIN; + return -1; + } + + // Error occured. + return -1; +} + +zmq::fd_t zmq::vtcp_connecter_t::connect () +{ + int rc = vtcp_acceptc (s, subport); + if (rc != 0) { + int err = errno; + close (); + errno = err; + return retired_fd; + } + + fd_t result = s; + s = retired_fd; + return result; +} + +int zmq::vtcp_connecter_t::close () +{ + zmq_assert (s != retired_fd); + int rc = ::close (s); + if (rc != 0) + return -1; + s = retired_fd; + return 0; +} + +#endif + diff --git a/src/vtcp_connecter.hpp b/src/vtcp_connecter.hpp new file mode 100644 index 0000000..61a592f --- /dev/null +++ b/src/vtcp_connecter.hpp @@ -0,0 +1,121 @@ +/* + 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 <http://www.gnu.org/licenses/>. +*/ + +#ifndef __VTCP_CONNECTER_HPP_INCLUDED__ +#define __VTCP_CONNECTER_HPP_INCLUDED__ + +#include "platform.hpp" + +#if defined ZMQ_HAVE_VTCP + +#include <vtcp.h> + +#include "fd.hpp" +#include "ip.hpp" +#include "own.hpp" +#include "io_object.hpp" +#include "stdint.hpp" + +namespace zmq +{ + + class vtcp_connecter_t : public own_t, public io_object_t + { + public: + + // If 'delay' is true connecter first waits for a while, then starts + // connection process. + vtcp_connecter_t (class io_thread_t *io_thread_, + class session_t *session_, const options_t &options_, + const char *address_, bool delay_); + ~vtcp_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; + vtcp_subport_t subport; + + // 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; + + vtcp_connecter_t (const vtcp_connecter_t&); + const vtcp_connecter_t &operator = (const vtcp_connecter_t&); + }; + +} + +#endif + +#endif diff --git a/src/vtcp_listener.cpp b/src/vtcp_listener.cpp index 31fb9ac..60fad8c 100644 --- a/src/vtcp_listener.cpp +++ b/src/vtcp_listener.cpp @@ -50,15 +50,20 @@ zmq::vtcp_listener_t::~vtcp_listener_t () int zmq::vtcp_listener_t::set_address (const char *addr_) { - // Find the '.' at end that separates NIC name from service. + // VTCP doesn't allow for binding to a specific interface. Connection + // string has to begin with *: (INADDR_ANY). + if (strlen (addr_) < 2 || addr_ [0] != '*' || addr_ [1] != ':') { + errno = EADDRNOTAVAIL; + return -1; + } + + // Parse port and subport. const char *delimiter = strrchr (addr_, '.'); if (!delimiter) { errno = EINVAL; return -1; } - - // Parse port and subport. - std::string port_str (addr_, delimiter - addr_); + std::string port_str (addr_ + 2, delimiter - addr_ - 2); std::string subport_str (delimiter + 1); uint16_t port = (uint16_t) atoi (port_str.c_str ()); uint32_t subport = (uint32_t) atoi (subport_str.c_str ()); |