diff options
| -rw-r--r-- | src/Makefile.am | 2 | ||||
| -rw-r--r-- | src/io_object.cpp | 13 | ||||
| -rw-r--r-- | src/io_object.hpp | 18 | ||||
| -rw-r--r-- | src/socket_base.cpp | 12 | ||||
| -rw-r--r-- | src/tcp_connecter.cpp | 24 | ||||
| -rw-r--r-- | src/tcp_connecter.hpp | 9 | ||||
| -rw-r--r-- | src/zmq_connecter.cpp | 110 | ||||
| -rw-r--r-- | src/zmq_connecter.hpp | 70 | ||||
| -rw-r--r-- | src/zmq_listener.cpp | 9 | ||||
| -rw-r--r-- | src/zmq_listener.hpp | 5 |
10 files changed, 239 insertions, 33 deletions
diff --git a/src/Makefile.am b/src/Makefile.am index 674f307..5d51725 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -44,6 +44,7 @@ libzmq_la_SOURCES = \ ypipe.hpp \ ypollset.hpp \ yqueue.hpp \ + zmq_connecter.hpp \ zmq_engine.hpp \ zmq_listener.hpp \ app_thread.cpp \ @@ -67,6 +68,7 @@ libzmq_la_SOURCES = \ uuid.cpp \ ypollset.cpp \ zmq.cpp \ + zmq_connecter.cpp \ zmq_engine.cpp \ zmq_listener.cpp diff --git a/src/io_object.cpp b/src/io_object.cpp index b7c70d4..d8cc1c0 100644 --- a/src/io_object.cpp +++ b/src/io_object.cpp @@ -52,9 +52,9 @@ void zmq::io_object_t::process_plug () plugged_in = true; } -zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_, i_poll_events *events_) +zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_) { - return poller->add_fd (fd_, events_); + return poller->add_fd (fd_, this); } void zmq::io_object_t::rm_fd (handle_t handle_) @@ -82,14 +82,14 @@ void zmq::io_object_t::reset_pollout (handle_t handle_) poller->reset_pollout (handle_); } -void zmq::io_object_t::add_timer (i_poll_events *events_) +void zmq::io_object_t::add_timer () { - poller->add_timer (events_); + poller->add_timer (this); } -void zmq::io_object_t::cancel_timer (i_poll_events *events_) +void zmq::io_object_t::cancel_timer () { - poller->cancel_timer (events_); + poller->cancel_timer (this); } void zmq::io_object_t::in_event () @@ -126,5 +126,6 @@ void zmq::io_object_t::process_term () // Otherwise, destroy the object and acknowledge the termination // straight away. send_term_ack (owner); + process_unplug (); delete this; } diff --git a/src/io_object.hpp b/src/io_object.hpp index 2029bfb..4f323ad 100644 --- a/src/io_object.hpp +++ b/src/io_object.hpp @@ -51,15 +51,20 @@ namespace zmq // handler. void process_plug (); + // io_object_t defines a new handler used to disconnect the object + // from the poller object. Implement the handlen in the derived + // classes to ensure sane cleanup. + virtual void process_unplug () = 0; + // Methods to access underlying poller object. - handle_t add_fd (fd_t fd_, struct i_poll_events *events_); + handle_t add_fd (fd_t fd_); void rm_fd (handle_t handle_); void set_pollin (handle_t handle_); void reset_pollin (handle_t handle_); void set_pollout (handle_t handle_); void reset_pollout (handle_t handle_); - void add_timer (struct i_poll_events *events_); - void cancel_timer (struct i_poll_events *events_); + void add_timer (); + void cancel_timer (); // i_poll_events interface implementation. void in_event (); @@ -70,12 +75,11 @@ namespace zmq // it when it's being closed. object_t *owner; - // Set to true when object is plugged in. It's responsibility - // of derived object to set the property after the feat. - bool plugged_in; - private: + // Set to true when object is plugged in. + bool plugged_in; + // Set to true when object was terminated before it was plugged in. // In such case destruction is delayed till 'plug' command arrives. bool terminated; diff --git a/src/socket_base.cpp b/src/socket_base.cpp index ac3b4b9..c179a93 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -25,6 +25,7 @@ #include "app_thread.hpp" #include "err.hpp" #include "zmq_listener.hpp" +#include "zmq_connecter.hpp" #include "io_thread.hpp" #include "config.hpp" @@ -127,7 +128,6 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_, int zmq::socket_base_t::bind (const char *addr_) { - // TODO: The taskset should be taken from socket options. zmq_listener_t *listener = new zmq_listener_t (choose_io_thread (affinity), this); int rc = listener->set_address (addr_); @@ -141,7 +141,15 @@ int zmq::socket_base_t::bind (const char *addr_) int zmq::socket_base_t::connect (const char *addr_) { - zmq_assert (false); + zmq_connecter_t *connecter = + new zmq_connecter_t (choose_io_thread (affinity), this); + int rc = connecter->set_address (addr_); + if (rc != 0) + return -1; + + send_plug (connecter); + send_own (this, connecter); + return 0; } int zmq::socket_base_t::subscribe (const char *criteria_) diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp index 7d29019..23f3741 100644 --- a/src/tcp_connecter.cpp +++ b/src/tcp_connecter.cpp @@ -17,6 +17,8 @@ along with this program. If not, see <http://www.gnu.org/licenses/>. */ +#include <string> + #include "tcp_connecter.hpp" #include "platform.hpp" #include "ip.hpp" @@ -40,6 +42,7 @@ zmq::tcp_connecter_t::tcp_connecter_t () : s (retired_fd) { + memset (&addr, 0, sizeof (addr)); } zmq::tcp_connecter_t::~tcp_connecter_t () @@ -48,15 +51,15 @@ zmq::tcp_connecter_t::~tcp_connecter_t () close (); } -int zmq::tcp_connecter_t::open (const char *addr_) +int zmq::tcp_connecter_t::set_address (const char *addr_) { - zmq_assert (s == retired_fd); - // Convert the hostname into sockaddr_in structure. - sockaddr_in address; - int rc = resolve_ip_hostname (&address, addr_); - if (rc != 0) - return -1; + return resolve_ip_hostname (&addr, addr_); +} + +int zmq::tcp_connecter_t::open () +{ + zmq_assert (s == retired_fd); // Create the socket. s = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP); @@ -67,7 +70,7 @@ int zmq::tcp_connecter_t::open (const char *addr_) int flags = fcntl (s, F_GETFL, 0); if (flags == -1) flags = 0; - rc = fcntl (s, F_SETFL, flags | O_NONBLOCK); + int rc = fcntl (s, F_SETFL, flags | O_NONBLOCK); errno_assert (rc != -1); // Disable Nagle's algorithm. @@ -83,7 +86,7 @@ int zmq::tcp_connecter_t::open (const char *addr_) #endif // Connect to the remote peer. - rc = ::connect (s, (sockaddr*) &address, sizeof address); + rc = ::connect (s, (sockaddr*) &addr, sizeof (addr)); // Connect was successfull immediately. if (rc == 0) @@ -91,7 +94,7 @@ int zmq::tcp_connecter_t::open (const char *addr_) // Asynchronous connect was launched. if (rc == -1 && errno == EINPROGRESS) - return 1; + return -1; // Error occured. int err = errno; @@ -125,7 +128,6 @@ zmq::fd_t zmq::tcp_connecter_t::connect () if (rc == -1) err = errno; if (err != 0) { - close (); errno = err; return retired_fd; } diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp index aa1ef05..bb358fd 100644 --- a/src/tcp_connecter.hpp +++ b/src/tcp_connecter.hpp @@ -21,6 +21,7 @@ #define __ZMQ_TCP_CONNECTER_HPP_INCLUDED__ #include "fd.hpp" +#include "ip.hpp" namespace zmq { @@ -34,11 +35,14 @@ namespace zmq tcp_connecter_t (); ~tcp_connecter_t (); + // Set IP address/port to connect to. + int set_address (const char *addr_); + // Open TCP connecting socket. Address is in // <hostname>:<port-number> format. Returns -1 in case of error, // 0 if connect was successfull immediately and 1 if async connect // was launched. - int open (const char *addr_); + int open (); // Close the connecting socket. int close (); @@ -53,6 +57,9 @@ namespace zmq private: + // Address to connect to. + sockaddr_in addr; + // Underlying socket. fd_t s; diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp new file mode 100644 index 0000000..513508d --- /dev/null +++ b/src/zmq_connecter.cpp @@ -0,0 +1,110 @@ +/* + Copyright (c) 2007-2009 FastMQ Inc. + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU 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 + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "zmq_connecter.hpp" +#include "err.hpp" + +zmq::zmq_connecter_t::zmq_connecter_t (io_thread_t *parent_, object_t *owner_) : + io_object_t (parent_, owner_), + waiting (false) +{ +} + +zmq::zmq_connecter_t::~zmq_connecter_t () +{ +} + +int zmq::zmq_connecter_t::set_address (const char *addr_) +{ + return tcp_connecter.set_address (addr_); +} + +void zmq::zmq_connecter_t::process_plug () +{ + start_connecting (); + io_object_t::process_plug (); +} + +void zmq::zmq_connecter_t::process_unplug () +{ + if (!waiting) + rm_fd (handle); +} + +void zmq::zmq_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::zmq_connecter_t::out_event () +{ + fd_t fd = tcp_connecter.connect (); + + // If there was error during the connecting, close the socket and wait + // for a while before trying to reconnect. + if (fd == retired_fd) { + rm_fd (handle); + tcp_connecter.close (); + waiting = true; + add_timer (); + return; + } + + zmq_assert (false); + +/* + object_t *engine = new zmq_engine_t (choose_io_thread (0), owner); + send_plug (engine); + send_own (owner, engine); +*/ +} + +void zmq::zmq_connecter_t::timer_event () +{ + // Reconnect period have elapsed. + waiting = false; + start_connecting (); +} + +void zmq::zmq_connecter_t::start_connecting () +{ + // Open the connecting socket. + int rc = tcp_connecter.open (); + + // Connect may succeed in synchronous manner. + if (rc == 0) { + out_event (); + return; + } + + // Connection establishment may be dealyed. Poll for its completion. + else if (rc == -1 && errno == EINPROGRESS) { + handle = add_fd (tcp_connecter.get_fd ()); + set_pollout (handle); + return; + } + + // If none of the above is true, synchronous error occured. + // Wait for a while and retry. + waiting = true; + add_timer (); +} diff --git a/src/zmq_connecter.hpp b/src/zmq_connecter.hpp new file mode 100644 index 0000000..d346396 --- /dev/null +++ b/src/zmq_connecter.hpp @@ -0,0 +1,70 @@ +/* + Copyright (c) 2007-2009 FastMQ Inc. + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU 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 + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#ifndef __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__ +#define __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__ + +#include "io_object.hpp" +#include "tcp_connecter.hpp" + +namespace zmq +{ + + class zmq_connecter_t : public io_object_t + { + public: + + zmq_connecter_t (class io_thread_t *parent_, object_t *owner_); + + // Set IP address to connect to. + int set_address (const char *addr_); + + private: + + ~zmq_connecter_t (); + + // Handlers for incoming commands. + void process_plug (); + void process_unplug (); + + // Handlers for I/O events. + void in_event (); + void out_event (); + void timer_event (); + + // Internal function to start the actual connection establishment. + void start_connecting (); + + // Actual connecting socket. + tcp_connecter_t tcp_connecter; + + // Handle corresponding to the listening socket. + handle_t handle; + + // True, if we are waiting for a period of time before trying to + // reconnect. + bool waiting; + + zmq_connecter_t (const zmq_connecter_t&); + void operator = (const zmq_connecter_t&); + }; + +} + +#endif diff --git a/src/zmq_listener.cpp b/src/zmq_listener.cpp index 9c9bbe9..463a1dd 100644 --- a/src/zmq_listener.cpp +++ b/src/zmq_listener.cpp @@ -29,8 +29,6 @@ zmq::zmq_listener_t::zmq_listener_t (io_thread_t *parent_, object_t *owner_) : zmq::zmq_listener_t::~zmq_listener_t () { - if (plugged_in) - rm_fd (handle); } int zmq::zmq_listener_t::set_address (const char *addr_) @@ -45,12 +43,17 @@ void zmq::zmq_listener_t::process_plug () zmq_assert (rc == 0); // Start polling for incoming connections. - handle = add_fd (tcp_listener.get_fd (), this); + handle = add_fd (tcp_listener.get_fd ()); set_pollin (handle); io_object_t::process_plug (); } +void zmq::zmq_listener_t::process_unplug () +{ + rm_fd (handle); +} + void zmq::zmq_listener_t::in_event () { fd_t fd = tcp_listener.accept (); diff --git a/src/zmq_listener.hpp b/src/zmq_listener.hpp index 74d42e1..ef252b3 100644 --- a/src/zmq_listener.hpp +++ b/src/zmq_listener.hpp @@ -20,8 +20,6 @@ #ifndef __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__ #define __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__ -#include <string> - #include "io_object.hpp" #include "tcp_listener.hpp" @@ -43,8 +41,9 @@ namespace zmq // Handlers for incoming commands. void process_plug (); + void process_unplug (); - // Handle I/O events. + // Handlers for I/O events. void in_event (); // Actual listening socket. |