From dbc42ee9af098183714845a75139997dcc34c7a0 Mon Sep 17 00:00:00 2001 From: Erik Hugne Date: Tue, 29 Jan 2013 14:16:38 +0100 Subject: [PATCH] zmq: add support for TIPC transport A ZeroMQ application can opt for TIPC based sockets using the TIPC port name format: zmq_bind(sb, "tipc://{type,lower,upper}"); zmq_connect(sc, "tipc://{type,inst}"); 'type' is the service ID, and 'lower/upper' can be used for service partitioning or basic load balancing. ZeroMQ TIPC transport requires a kernel >= 3.8 (nonblocking connect support for TIPC). Signed-off-by: Erik Hugne --- src/Makefile.am | 8 +- src/address.cpp | 17 +++ src/address.hpp | 6 + src/session_base.cpp | 11 ++ src/socket_base.cpp | 43 ++++++- src/tipc_address.cpp | 105 +++++++++++++++++ src/tipc_address.hpp | 67 +++++++++++ src/tipc_connecter.cpp | 257 +++++++++++++++++++++++++++++++++++++++++ src/tipc_connecter.hpp | 127 ++++++++++++++++++++ src/tipc_listener.cpp | 178 ++++++++++++++++++++++++++++ src/tipc_listener.hpp | 96 +++++++++++++++ 11 files changed, 912 insertions(+), 3 deletions(-) create mode 100644 src/tipc_address.cpp create mode 100644 src/tipc_address.hpp create mode 100644 src/tipc_connecter.cpp create mode 100644 src/tipc_connecter.hpp create mode 100644 src/tipc_listener.cpp create mode 100644 src/tipc_listener.hpp diff --git a/src/Makefile.am b/src/Makefile.am index 340c997b..3df44a40 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -146,7 +146,13 @@ libzmq_la_SOURCES = \ v1_protocol.hpp \ xsub.cpp \ zmq.cpp \ - zmq_utils.cpp + zmq_utils.cpp \ + tipc_address.cpp \ + tipc_address.hpp \ + tipc_listener.cpp \ + tipc_listener.hpp \ + tipc_connecter.cpp\ + tipc_connecter.hpp if ON_MINGW libzmq_la_LDFLAGS = -no-undefined -avoid-version -version-info @LTVER@ @LIBZMQ_EXTRA_LDFLAGS@ diff --git a/src/address.cpp b/src/address.cpp index 6dc86dd1..a38c9870 100644 --- a/src/address.cpp +++ b/src/address.cpp @@ -18,10 +18,12 @@ along with this program. If not, see . */ +#include "platform.hpp" #include "address.hpp" #include "err.hpp" #include "tcp_address.hpp" #include "ipc_address.hpp" +#include "tipc_address.hpp" #include #include @@ -50,6 +52,14 @@ zmq::address_t::~address_t () } } #endif +#if defined ZMQ_HAVE_LINUX + else if (protocol == "tipc") { + if (resolved.tipc_addr) { + delete resolved.tipc_addr; + resolved.tipc_addr = 0; + } + } +#endif } int zmq::address_t::to_string (std::string &addr_) const @@ -66,6 +76,13 @@ int zmq::address_t::to_string (std::string &addr_) const } } #endif +#if defined ZMQ_HAVE_LINUX + else if (protocol == "tipc") { + if (resolved.tipc_addr) { + return resolved.tipc_addr->to_string(addr_); + } + } +#endif if (!protocol.empty () && !address.empty ()) { std::stringstream s; diff --git a/src/address.hpp b/src/address.hpp index a8a4765d..2156ccf3 100644 --- a/src/address.hpp +++ b/src/address.hpp @@ -28,6 +28,9 @@ namespace zmq class tcp_address_t; #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS class ipc_address_t; +#endif +#if defined ZMQ_HAVE_LINUX + class tipc_address_t; #endif struct address_t { address_t (const std::string &protocol_, const std::string &address_); @@ -42,6 +45,9 @@ namespace zmq tcp_address_t *tcp_addr; #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS ipc_address_t *ipc_addr; +#endif +#if defined ZMQ_HAVE_LINUX + tipc_address_t *tipc_addr; #endif } resolved; diff --git a/src/session_base.cpp b/src/session_base.cpp index b2676570..dc85e6e6 100644 --- a/src/session_base.cpp +++ b/src/session_base.cpp @@ -27,6 +27,7 @@ #include "err.hpp" #include "pipe.hpp" #include "likely.hpp" +#include "tipc_connecter.hpp" #include "tcp_connecter.hpp" #include "ipc_connecter.hpp" #include "pgm_sender.hpp" @@ -456,6 +457,16 @@ void zmq::session_base_t::start_connecting (bool wait_) } #endif +#if defined ZMQ_HAVE_LINUX + if (addr->protocol == "tipc") { + tipc_connecter_t *connecter = new (std::nothrow) tipc_connecter_t ( + io_thread, this, options, addr, 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/socket_base.cpp b/src/socket_base.cpp index 7fe40089..4fea5448 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -55,6 +55,8 @@ #include "address.hpp" #include "ipc_address.hpp" #include "tcp_address.hpp" +#include "tipc_address.hpp" +#include "tipc_listener.hpp" #ifdef ZMQ_HAVE_OPENPGM #include "pgm_socket.hpp" #endif @@ -182,7 +184,7 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_) { // First check out whether the protcol is something we are aware of. if (protocol_ != "inproc" && protocol_ != "ipc" && protocol_ != "tcp" && - protocol_ != "pgm" && protocol_ != "epgm") { + protocol_ != "pgm" && protocol_ != "epgm" && protocol_ != "tipc") { errno = EPROTONOSUPPORT; return -1; } @@ -204,7 +206,13 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_) return -1; } #endif - + // TIPC transport is only available on Linux. +#if !defined ZMQ_HAVE_LINUX + if (protocol_ = "tipc") { + errno = EPROTONOSUPPORT; + return -1; + } +#endif // Check whether socket type and transport protocol match. // Specifically, multicast protocols can't be combined with // bi-directional messaging patterns (socket types). @@ -387,6 +395,25 @@ int zmq::socket_base_t::bind (const char *addr_) return 0; } #endif +#if defined ZMQ_HAVE_LINUX + if (protocol == "tipc") { + tipc_listener_t *listener = new (std::nothrow) tipc_listener_t ( + io_thread, this, options); + alloc_assert (listener); + int rc = listener->set_address (address.c_str ()); + if (rc != 0) { + delete listener; + event_bind_failed (address, zmq_errno()); + return -1; + } + + // Save last endpoint URI + listener->get_address (options.last_endpoint); + + add_endpoint (addr_, (own_t *) listener); + return 0; + } +#endif zmq_assert (false); return -1; @@ -517,6 +544,18 @@ int zmq::socket_base_t::connect (const char *addr_) } } #endif +#if defined ZMQ_HAVE_LINUX + else + if (protocol == "tipc") { + paddr->resolved.tipc_addr = new (std::nothrow) tipc_address_t (); + alloc_assert (paddr->resolved.tipc_addr); + int rc = paddr->resolved.tipc_addr->resolve (address.c_str()); + if (rc != 0) { + delete paddr; + return -1; + } + } +#endif #ifdef ZMQ_HAVE_OPENPGM if (protocol == "pgm" || protocol == "epgm") { struct pgm_addrinfo_t *res = NULL; diff --git a/src/tipc_address.cpp b/src/tipc_address.cpp new file mode 100644 index 00000000..555b1917 --- /dev/null +++ b/src/tipc_address.cpp @@ -0,0 +1,105 @@ +/* + Copyright (c) 2013 Ericsson AB + + 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 "tipc_address.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include "err.hpp" + +#include +#include + +zmq::tipc_address_t::tipc_address_t () +{ + memset (&address, 0, sizeof (address)); +} + +zmq::tipc_address_t::tipc_address_t (const sockaddr *sa, socklen_t sa_len) +{ + zmq_assert(sa && sa_len > 0); + + memset (&address, 0, sizeof (address)); + if (sa->sa_family == AF_TIPC) { + memcpy(&address, sa, sa_len); + } +} + +zmq::tipc_address_t::~tipc_address_t () +{ +} + +int zmq::tipc_address_t::resolve (const char *name) +{ + int res; + unsigned int type = 0; + unsigned int lower = 0; + unsigned int upper = 0; + + res = sscanf(name, "{%u,%u,%u}", &type, &lower, &upper); + if (res == 3) + goto nameseq; + else if (res == 2 && type > TIPC_RESERVED_TYPES) { + address.family = AF_TIPC; + address.addrtype = TIPC_ADDR_NAME; + address.addr.name.name.type = type; + address.addr.name.name.instance = lower; + address.scope = 0; + return 0; + } + else + return EINVAL; +nameseq: + if (type < TIPC_RESERVED_TYPES || upper < lower) + return EINVAL; + address.family = AF_TIPC; + address.addrtype = TIPC_ADDR_NAMESEQ; + address.addr.nameseq.type = type; + address.addr.nameseq.lower = lower; + address.addr.nameseq.upper = upper; + address.scope = TIPC_ZONE_SCOPE; + + return 0; +} + +int zmq::tipc_address_t::to_string (std::string &addr_) +{ + if (address.family != AF_TIPC) { + addr_.clear (); + return -1; + } + std::stringstream s; + s << "tipc://" << "{" << address.addr.nameseq.type; + s << ", " << address.addr.nameseq.lower; + s << ", " << address.addr.nameseq.upper << "}"; + addr_ = s.str (); + return 0; +} + +const sockaddr *zmq::tipc_address_t::addr () const +{ + return (sockaddr*) &address; +} + +socklen_t zmq::tipc_address_t::addrlen () const +{ + return (socklen_t) sizeof (address); +} + +#endif diff --git a/src/tipc_address.hpp b/src/tipc_address.hpp new file mode 100644 index 00000000..d87dc43c --- /dev/null +++ b/src/tipc_address.hpp @@ -0,0 +1,67 @@ +/* + Copyright (c) 2011 250bpm s.r.o. + Copyright (c) 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_TIPC_ADDRESS_HPP_INCLUDED__ +#define __ZMQ_TIPC_ADDRESS_HPP_INCLUDED__ + +#include + +#include "platform.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include +#include + +namespace zmq +{ + + class tipc_address_t + { + public: + + tipc_address_t (); + tipc_address_t (const sockaddr *sa, socklen_t sa_len); + ~tipc_address_t (); + + // This function sets up the address "{type, lower, upper}" for TIPC transport + int resolve (const char* name); + + // The opposite to resolve() + int to_string (std::string &addr_); + + const sockaddr *addr () const; + socklen_t addrlen () const; + + private: + + struct sockaddr_tipc address; + + tipc_address_t (const tipc_address_t&); + const tipc_address_t &operator = (const tipc_address_t&); + }; + +} + +#endif + +#endif + + diff --git a/src/tipc_connecter.cpp b/src/tipc_connecter.cpp new file mode 100644 index 00000000..7c927b2f --- /dev/null +++ b/src/tipc_connecter.cpp @@ -0,0 +1,257 @@ +/* + Copyright (c) 2013 Ericsson AB + + 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 "tipc_connecter.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include +#include + +#include "stream_engine.hpp" +#include "io_thread.hpp" +#include "platform.hpp" +#include "random.hpp" +#include "err.hpp" +#include "ip.hpp" +#include "address.hpp" +#include "tipc_address.hpp" +#include "session_base.hpp" + +#include +#include +#include + +zmq::tipc_connecter_t::tipc_connecter_t (class io_thread_t *io_thread_, + class session_base_t *session_, const options_t &options_, + const address_t *addr_, bool delayed_start_) : + own_t (io_thread_, options_), + io_object_t (io_thread_), + addr (addr_), + s (retired_fd), + handle_valid (false), + delayed_start (delayed_start_), + timer_started (false), + session (session_), + current_reconnect_ivl(options.reconnect_ivl) +{ + zmq_assert (addr); + zmq_assert (addr->protocol == "tipc"); + addr->to_string (endpoint); + socket = session-> get_socket(); +} + +zmq::tipc_connecter_t::~tipc_connecter_t () +{ + zmq_assert (!timer_started); + zmq_assert (!handle_valid); + zmq_assert (s == retired_fd); +} + +void zmq::tipc_connecter_t::process_plug () +{ + if (delayed_start) + add_reconnect_timer (); + else + start_connecting (); +} + +void zmq::tipc_connecter_t::process_term (int linger_) +{ + if (timer_started) { + cancel_timer (reconnect_timer_id); + timer_started = false; + } + + if (handle_valid) { + rm_fd (handle); + handle_valid = false; + } + + if (s != retired_fd) + close (); + + own_t::process_term (linger_); +} + +void zmq::tipc_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::tipc_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 (); + add_reconnect_timer(); + return; + } + // Create the engine object for this connection. + stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint); + alloc_assert (engine); + + // Attach the engine to the corresponding session object. + send_attach (session, engine); + + // Shut the connecter down. + terminate (); + + socket->event_connected (endpoint, fd); +} + +void zmq::tipc_connecter_t::timer_event (int id_) +{ + zmq_assert (id_ == reconnect_timer_id); + timer_started = false; + start_connecting (); +} + +void zmq::tipc_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 (); + } + + // Connection establishment may be delayed. Poll for its completion. + else + if (rc == -1 && errno == EINPROGRESS) { + handle = add_fd (s); + handle_valid = true; + set_pollout (handle); + socket->event_connect_delayed (endpoint, zmq_errno()); + } + + // Handle any other error condition by eventual reconnect. + else { + if (s != retired_fd) + close (); + add_reconnect_timer (); + } +} + +void zmq::tipc_connecter_t::add_reconnect_timer() +{ + int rc_ivl = get_new_reconnect_ivl(); + add_timer (rc_ivl, reconnect_timer_id); + socket->event_connect_retried (endpoint, rc_ivl); + timer_started = true; +} + +int zmq::tipc_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; +} + +int zmq::tipc_connecter_t::open () +{ + zmq_assert (s == retired_fd); + + // Create the socket. + s = open_socket (AF_TIPC, SOCK_STREAM, 0); + if (s == -1) + return -1; + + // Set the non-blocking flag. + unblock_socket (s); + // Connect to the remote peer. + int rc = ::connect ( + s, addr->resolved.tipc_addr->addr (), + addr->resolved.tipc_addr->addrlen ()); + + // Connect was successfull immediately. + if (rc == 0) + return 0; + + // Translate other error codes indicating asynchronous connect has been + // launched to a uniform EINPROGRESS. + if (rc == -1 && errno == EINTR) { + errno = EINPROGRESS; + return -1; + } + // Forward the error. + return -1; +} + +void zmq::tipc_connecter_t::close () +{ + zmq_assert (s != retired_fd); + int rc = ::close (s); + errno_assert (rc == 0); + socket->event_closed (endpoint, s); + s = retired_fd; +} + +zmq::fd_t zmq::tipc_connecter_t::connect () +{ + // Following code should handle both Berkeley-derived socket + // implementations and Solaris. + int err = 0; + socklen_t len = sizeof (err); + + 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/tipc_connecter.hpp b/src/tipc_connecter.hpp new file mode 100644 index 00000000..5e4a3121 --- /dev/null +++ b/src/tipc_connecter.hpp @@ -0,0 +1,127 @@ +/* + Copyright (c) 2013 Ericsson AB + + 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 __TIPC_CONNECTER_HPP_INCLUDED__ +#define __TIPC_CONNECTER_HPP_INCLUDED__ + +#include "platform.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include "fd.hpp" +#include "own.hpp" +#include "stdint.hpp" +#include "io_object.hpp" + +namespace zmq +{ + + class io_thread_t; + class session_base_t; + struct address_t; + + class tipc_connecter_t : public own_t, public io_object_t + { + public: + + // If 'delayed_start' is true connecter first waits for a while, + // then starts connection process. + tipc_connecter_t (zmq::io_thread_t *io_thread_, + zmq::session_base_t *session_, const options_t &options_, + const address_t *addr_, bool delayed_start_); + ~tipc_connecter_t (); + + private: + + // ID of the timer used to delay the reconnection. + enum {reconnect_timer_id = 1}; + + // Handlers for incoming commands. + void process_plug (); + void process_term (int linger_); + + // 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 (); + + // Open IPC connecting socket. Returns -1 in case of error, + // 0 if connect was successfull immediately. Returns -1 with + // EAGAIN errno if async connect was launched. + int open (); + + // Close the connecting socket. + void close (); + + // Get the file descriptor of newly created connection. Returns + // retired_fd if the connection was unsuccessfull. + fd_t connect (); + + // Address to connect to. Owned by session_base_t. + const address_t *addr; + + // 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. + const bool delayed_start; + + // True iff a timer has been started. + bool timer_started; + + // Reference to the session we belong to. + zmq::session_base_t *session; + + // Current reconnect ivl, updated for backoff strategy + int current_reconnect_ivl; + + // String representation of endpoint to connect to + std::string endpoint; + + // Socket + zmq::socket_base_t *socket; + + tipc_connecter_t (const tipc_connecter_t&); + const tipc_connecter_t &operator = (const tipc_connecter_t&); + }; + +} + +#endif + +#endif + diff --git a/src/tipc_listener.cpp b/src/tipc_listener.cpp new file mode 100644 index 00000000..0b1a1b09 --- /dev/null +++ b/src/tipc_listener.cpp @@ -0,0 +1,178 @@ +/* + Copyright (c) 2011 250bpm s.r.o. + Copyright (c) 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 "tipc_listener.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include + +#include + +#include "stream_engine.hpp" +#include "tipc_address.hpp" +#include "io_thread.hpp" +#include "session_base.hpp" +#include "config.hpp" +#include "err.hpp" +#include "ip.hpp" +#include "socket_base.hpp" + +#include +#include +#include +#include + +zmq::tipc_listener_t::tipc_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_), + s (retired_fd), + socket (socket_) +{ +} + +zmq::tipc_listener_t::~tipc_listener_t () +{ + zmq_assert (s == retired_fd); +} + +void zmq::tipc_listener_t::process_plug () +{ + // Start polling for incoming connections. + handle = add_fd (s); + set_pollin (handle); +} + +void zmq::tipc_listener_t::process_term (int linger_) +{ + rm_fd (handle); + close (); + own_t::process_term (linger_); +} + +void zmq::tipc_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) { + socket->event_accept_failed (endpoint, zmq_errno()); + return; + } + + // Create the engine object for this connection. + stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint); + 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_base_t *session = session_base_t::create (io_thread, false, socket, + options, NULL); + errno_assert (session); + session->inc_seqnum (); + launch_child (session); + send_attach (session, engine, false); + socket->event_accepted (endpoint, fd); +} + +int zmq::tipc_listener_t::get_address (std::string &addr_) +{ + struct sockaddr_storage ss; + socklen_t sl = sizeof (ss); + + int rc = getsockname (s, (sockaddr *) &ss, &sl); + if (rc != 0) { + addr_.clear (); + return rc; + } + + tipc_address_t addr ((struct sockaddr *) &ss, sl); + return addr.to_string (addr_); +} + +int zmq::tipc_listener_t::set_address (const char *addr_) +{ + //convert str to address struct + int rc = address.resolve(addr_); + if (rc != 0) + return -1; + // Create a listening socket. + s = open_socket (AF_TIPC, SOCK_STREAM, 0); + if (s == -1) + return -1; + + address.to_string (endpoint); + + // Bind the socket to tipc name. + rc = bind (s, address.addr (), address.addrlen ()); + if (rc != 0) + goto error; + + // Listen for incomming connections. + rc = listen (s, options.backlog); + if (rc != 0) + goto error; + + socket->event_listening (endpoint, s); + return 0; + +error: + int err = errno; + close (); + errno = err; + return -1; +} + +void zmq::tipc_listener_t::close () +{ + zmq_assert (s != retired_fd); + int rc = ::close (s); + errno_assert (rc == 0); + s = retired_fd; + socket->event_closed (endpoint, s); +} + +zmq::fd_t zmq::tipc_listener_t::accept () +{ + // Accept one connection and deal with different failure modes. + // The situation where connection cannot be accepted due to insufficient + // resources is considered valid and treated by ignoring the connection. + struct sockaddr_storage ss = {}; + socklen_t ss_len = sizeof(ss); + + zmq_assert (s != retired_fd); + fd_t sock = ::accept (s, (struct sockaddr *) &ss, &ss_len); + if (sock == -1) { + errno_assert (errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || + errno == EINTR || errno == ECONNABORTED || errno == EPROTO || errno == EMFILE || + errno == ENFILE); + return retired_fd; + } + /*FIXME Accept filters?*/ + return sock; +} + +#endif diff --git a/src/tipc_listener.hpp b/src/tipc_listener.hpp new file mode 100644 index 00000000..d5d3e293 --- /dev/null +++ b/src/tipc_listener.hpp @@ -0,0 +1,96 @@ +/* + Copyright (c) 2013 Ericsson AB + + 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_TIPC_LISTENER_HPP_INCLUDED__ +#define __ZMQ_TIPC_LISTENER_HPP_INCLUDED__ + +#include "platform.hpp" + +#if defined ZMQ_HAVE_LINUX + +#include + +#include "fd.hpp" +#include "own.hpp" +#include "stdint.hpp" +#include "io_object.hpp" +#include "tipc_address.hpp" + +namespace zmq +{ + + class io_thread_t; + class socket_base_t; + + class tipc_listener_t : public own_t, public io_object_t + { + public: + + tipc_listener_t (zmq::io_thread_t *io_thread_, + zmq::socket_base_t *socket_, const options_t &options_); + ~tipc_listener_t (); + + // Set address to listen on. + int set_address (const char *addr_); + + // Get the bound address for use with wildcards + int get_address (std::string &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. + void 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 + tipc_address_t address; + + // Underlying socket. + fd_t s; + + // Handle corresponding to the listening socket. + handle_t handle; + + // Socket the listerner belongs to. + zmq::socket_base_t *socket; + + // String representation of endpoint to bind to + std::string endpoint; + + tipc_listener_t (const tipc_listener_t&); + const tipc_listener_t &operator = (const tipc_listener_t&); + }; + +} + +#endif + +#endif +