diff options
| -rw-r--r-- | include/zmq.h | 28 | ||||
| -rw-r--r-- | include/zmq.hpp | 14 | ||||
| -rw-r--r-- | src/Makefile.am | 2 | ||||
| -rw-r--r-- | src/epoll.cpp | 4 | ||||
| -rw-r--r-- | src/i_api.hpp | 6 | ||||
| -rw-r--r-- | src/io_object.cpp | 45 | ||||
| -rw-r--r-- | src/io_object.hpp | 23 | ||||
| -rw-r--r-- | src/io_thread.hpp | 2 | ||||
| -rw-r--r-- | src/socket_base.cpp | 33 | ||||
| -rw-r--r-- | src/socket_base.hpp | 9 | ||||
| -rw-r--r-- | src/zmq.cpp | 13 | ||||
| -rw-r--r-- | src/zmq_engine.cpp | 35 | ||||
| -rw-r--r-- | src/zmq_engine.hpp | 47 | ||||
| -rw-r--r-- | src/zmq_listener.cpp | 9 | ||||
| -rw-r--r-- | src/zmq_listener.hpp | 5 |
15 files changed, 222 insertions, 53 deletions
diff --git a/include/zmq.h b/include/zmq.h index 0fbc142..63c8434 100644 --- a/include/zmq.h +++ b/include/zmq.h @@ -42,6 +42,14 @@ extern "C" { #define ZMQ_DELIMITER 31 #define ZMQ_VSM 32 +// Socket options. +#define ZMQ_HWM 1 +#define ZMQ_LWM 2 +#define ZMQ_SWAP 3 +#define ZMQ_MASK 4 +#define ZMQ_AFFINITY 5 +#define ZMQ_SESSIONID 6 + // The operation should be performed in non-blocking mode. I.e. if it cannot // be processed immediately, error should be returned with errno set to EAGAIN. #define ZMQ_NOBLOCK 1 @@ -94,18 +102,6 @@ struct zmq_msg unsigned char vsm_data [ZMQ_MAX_VSM_SIZE]; }; -// TODO: Different options... -struct zmq_opts -{ - uint64_t hwm; - uint64_t lwm; - uint64_t swap; - uint64_t mask; - uint64_t taskset; - const char *identity; - const char *args; -}; - // Initialise an empty message (zero bytes long). ZMQ_EXPORT int zmq_msg_init (zmq_msg *msg); @@ -165,11 +161,15 @@ ZMQ_EXPORT void *zmq_socket (void *context, int type); // Close the socket. ZMQ_EXPORT int zmq_close (void *s); +// Sets an option on the socket. +ZMQ_EXPORT int zmq_setsockopt (void *s, int option_, void *optval_, + size_t optvallen_); + // Bind the socket to a particular address. -ZMQ_EXPORT int zmq_bind (void *s, const char *addr, zmq_opts *opts); +ZMQ_EXPORT int zmq_bind (void *s, const char *addr); // Connect the socket to a particular address. -ZMQ_EXPORT int zmq_connect (void *s, const char *addr, zmq_opts *opts); +ZMQ_EXPORT int zmq_connect (void *s, const char *addr); // Subscribe for the subset of messages identified by 'criteria' argument. ZMQ_EXPORT int zmq_subscribe (void *s, const char *criteria); diff --git a/include/zmq.hpp b/include/zmq.hpp index 004706b..0ba972b 100644 --- a/include/zmq.hpp +++ b/include/zmq.hpp @@ -230,9 +230,15 @@ namespace zmq assert (rc == 0); } - inline void bind (const char *addr_, zmq_opts *opts_ = NULL) + template <typename T> inline void setsockopt (int option_, T &value_) { - int rc = zmq_bind (ptr, addr_, opts_); + int rc = zmq_setsockopt (ptr, option_, (void*) &value_, sizeof (T)); + assert (rc == 0); + } + + inline void bind (const char *addr_) + { + int rc = zmq_bind (ptr, addr_); if (rc == -1) { assert (errno == EINVAL || errno == EADDRINUSE); if (errno == EINVAL) @@ -242,9 +248,9 @@ namespace zmq } } - inline void connect (const char *addr_, zmq_opts *opts_ = NULL) + inline void connect (const char *addr_) { - int rc = zmq_connect (ptr, addr_, opts_); + int rc = zmq_connect (ptr, addr_); if (rc == -1) { assert (errno == EINVAL || errno == EADDRINUSE); if (errno == EINVAL) diff --git a/src/Makefile.am b/src/Makefile.am index 47037a2..674f307 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -44,6 +44,7 @@ libzmq_la_SOURCES = \ ypipe.hpp \ ypollset.hpp \ yqueue.hpp \ + zmq_engine.hpp \ zmq_listener.hpp \ app_thread.cpp \ devpoll.cpp \ @@ -66,6 +67,7 @@ libzmq_la_SOURCES = \ uuid.cpp \ ypollset.cpp \ zmq.cpp \ + zmq_engine.cpp \ zmq_listener.cpp libzmq_la_LDFLAGS = -version-info 0:0:0 diff --git a/src/epoll.cpp b/src/epoll.cpp index 15278c6..5f3bc51 100644 --- a/src/epoll.cpp +++ b/src/epoll.cpp @@ -41,10 +41,12 @@ zmq::epoll_t::epoll_t () : zmq::epoll_t::~epoll_t () { + // Wait till the worker thread exits. + worker.stop (); + // Make sure there are no fds registered on shutdown. zmq_assert (load.get () == 0); - worker.stop (); close (epoll_fd); for (retired_t::iterator it = retired.begin (); it != retired.end (); it ++) delete *it; diff --git a/src/i_api.hpp b/src/i_api.hpp index 36afcea..e9bf1da 100644 --- a/src/i_api.hpp +++ b/src/i_api.hpp @@ -27,8 +27,10 @@ namespace zmq { virtual ~i_api () {} - virtual int bind (const char *addr_, struct zmq_opts *opts_) = 0; - virtual int connect (const char *addr_, struct zmq_opts *opts_) = 0; + virtual int setsockopt (int option_, void *optval_, + size_t optvallen_) = 0; + virtual int bind (const char *addr_) = 0; + virtual int connect (const char *addr_) = 0; virtual int subscribe (const char *criteria_) = 0; virtual int send (struct zmq_msg *msg_, int flags_) = 0; virtual int flush () = 0; diff --git a/src/io_object.cpp b/src/io_object.cpp index 41e4717..a4badd7 100644 --- a/src/io_object.cpp +++ b/src/io_object.cpp @@ -18,17 +18,60 @@ */ #include "io_object.hpp" +#include "io_thread.hpp" -zmq::io_object_t::io_object_t (object_t *parent_, object_t *owner_) : +zmq::io_object_t::io_object_t (io_thread_t *parent_, object_t *owner_) : object_t (parent_), owner (owner_) { + // Retrieve the poller from the thread we are running in. + poller = parent_->get_poller (); } zmq::io_object_t::~io_object_t () { } +zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_, i_poll_events *events_) +{ + return poller->add_fd (fd_, events_); +} + +void zmq::io_object_t::rm_fd (handle_t handle_) +{ + poller->rm_fd (handle_); +} + +void zmq::io_object_t::set_pollin (handle_t handle_) +{ + poller->set_pollin (handle_); +} + +void zmq::io_object_t::reset_pollin (handle_t handle_) +{ + poller->reset_pollin (handle_); +} + +void zmq::io_object_t::set_pollout (handle_t handle_) +{ + poller->set_pollout (handle_); +} + +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_) +{ + poller->add_timer (events_); +} + +void zmq::io_object_t::cancel_timer (i_poll_events *events_) +{ + poller->cancel_timer (events_); +} + void zmq::io_object_t::term () { send_term_req (owner, this); diff --git a/src/io_object.hpp b/src/io_object.hpp index 5ed1830..ddb4414 100644 --- a/src/io_object.hpp +++ b/src/io_object.hpp @@ -21,6 +21,7 @@ #define __ZMQ_IO_OBJECT_HPP_INCLUDED__ #include "object.hpp" +#include "i_poller.hpp" namespace zmq { @@ -31,7 +32,7 @@ namespace zmq // I/O object will live in the thread inherited from the parent. // However, it's lifetime is managed by the owner. - io_object_t (object_t *parent_, object_t *owner_); + io_object_t (class io_thread_t *parent_, object_t *owner_); protected: @@ -44,15 +45,27 @@ namespace zmq // of I/O object correctly. virtual ~io_object_t (); - private: - - // Handlers for incoming commands. - void process_term (); + // Methods to access underlying poller object. + handle_t add_fd (fd_t fd_, struct i_poll_events *events_); + 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_); // Socket owning this I/O object. It is responsible for destroying // it when it's being closed. object_t *owner; + private: + + struct i_poller *poller; + + // Handlers for incoming commands. + void process_term (); + io_object_t (const io_object_t&); void operator = (const io_object_t&); }; diff --git a/src/io_thread.hpp b/src/io_thread.hpp index 6f25627..f95880a 100644 --- a/src/io_thread.hpp +++ b/src/io_thread.hpp @@ -57,7 +57,7 @@ namespace zmq void out_event (); void timer_event (); - // ??? + // Used by io_objects to retrieve the assciated poller object. struct i_poller *get_poller (); // Command handlers. diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 3737410..026b317 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -46,8 +46,9 @@ zmq::socket_base_t::~socket_base_t () break; // Send termination request to all associated I/O objects. - for (io_objects_t::size_type i = 0; i != io_objects.size (); i++) - send_term (io_objects [i]); + for (io_objects_t::iterator it = io_objects.begin (); + it != io_objects.end (); it++) + send_term (*it); // Move the objects to the list of pending term acks. pending_term_acks += io_objects.size (); @@ -59,16 +60,23 @@ zmq::socket_base_t::~socket_base_t () } } -int zmq::socket_base_t::bind (const char *addr_, struct zmq_opts *opts_) +int zmq::socket_base_t::setsockopt (int option_, void *optval_, + size_t optvallen_) { - uint64_t taskset = opts_ ? opts_->taskset : 0; + zmq_assert (false); +} + +int zmq::socket_base_t::bind (const char *addr_) +{ + // TODO: The taskset should be taken from socket options. + uint64_t taskset = 0; object_t *listener = new zmq_listener_t (choose_io_thread (taskset), this); send_plug (listener); send_own (this, listener); return 0; } -int zmq::socket_base_t::connect (const char *addr_, struct zmq_opts *opts_) +int zmq::socket_base_t::connect (const char *addr_) { zmq_assert (false); } @@ -102,24 +110,23 @@ int zmq::socket_base_t::close () void zmq::socket_base_t::process_own (object_t *object_) { - io_objects.push_back (object_); + io_objects.insert (object_); } void zmq::socket_base_t::process_term_req (object_t *object_) { // If I/O object is well and alive ask it to terminate. - // TODO: Following find may produce an unacceptable jitter in - // C10K-style applications. If so, use set instead of vector. io_objects_t::iterator it = std::find (io_objects.begin (), io_objects.end (), object_); - if (it != io_objects.end ()) { - pending_term_acks++; - io_objects.erase (it); - send_term (object_); - } // If not found, we assume that termination request was already sent to // the object so we can sagely ignore the request. + if (it == io_objects.end ()) + return; + + pending_term_acks++; + io_objects.erase (it); + send_term (object_); } void zmq::socket_base_t::process_term_ack () diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 633f003..f028c18 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -20,7 +20,7 @@ #ifndef __ZMQ_SOCKET_BASE_HPP_INCLUDED__ #define __ZMQ_SOCKET_BASE_HPP_INCLUDED__ -#include <vector> +#include <set> #include "i_api.hpp" #include "object.hpp" @@ -36,8 +36,9 @@ namespace zmq ~socket_base_t (); // i_api interface implementation. - int bind (const char *addr_, struct zmq_opts *opts_); - int connect (const char *addr_, struct zmq_opts *opts_); + int setsockopt (int option_, void *optval_, size_t optvallen_); + int bind (const char *addr_); + int connect (const char *addr_); int subscribe (const char *criteria_); int send (struct zmq_msg *msg_, int flags_); int flush (); @@ -53,7 +54,7 @@ namespace zmq // List of all I/O objects owned by this socket. The socket is // responsible for deallocating them before it quits. - typedef std::vector <object_t*> io_objects_t; + typedef std::set <object_t*> io_objects_t; io_objects_t io_objects; // Number of I/O objects that were already asked to terminate diff --git a/src/zmq.cpp b/src/zmq.cpp index 149a7e2..1ea13bc 100644 --- a/src/zmq.cpp +++ b/src/zmq.cpp @@ -192,14 +192,19 @@ int zmq_close (void *s_) return 0; } -int zmq_bind (void *s_, const char *addr_, zmq_opts *opts_) +int zmq_setsockopt (void *s_, int option_, void *optval_, size_t optvallen_) { - return (((zmq::i_api*) s_)->bind (addr_, opts_)); + return (((zmq::i_api*) s_)->setsockopt (option_, optval_, optvallen_)); } -int zmq_connect (void *s_, const char *addr_, zmq_opts *opts_) +int zmq_bind (void *s_, const char *addr_) { - return (((zmq::i_api*) s_)->connect (addr_, opts_)); + return (((zmq::i_api*) s_)->bind (addr_)); +} + +int zmq_connect (void *s_, const char *addr_) +{ + return (((zmq::i_api*) s_)->connect (addr_)); } int zmq_subscribe (void *s_, const char *criteria_) diff --git a/src/zmq_engine.cpp b/src/zmq_engine.cpp new file mode 100644 index 0000000..3708c9a --- /dev/null +++ b/src/zmq_engine.cpp @@ -0,0 +1,35 @@ +/* + 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_engine.hpp" +#include "io_thread.hpp" + +zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, object_t *owner_) : + io_object_t (parent_, owner_) +{ +} + +zmq::zmq_engine_t::~zmq_engine_t () +{ +} + +void zmq::zmq_engine_t::process_plug () +{ +} + diff --git a/src/zmq_engine.hpp b/src/zmq_engine.hpp new file mode 100644 index 0000000..ad739c7 --- /dev/null +++ b/src/zmq_engine.hpp @@ -0,0 +1,47 @@ +/* + 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_ENGINE_HPP_INCLUDED__ +#define __ZMQ_ZMQ_ENGINE_HPP_INCLUDED__ + +#include "io_object.hpp" + +namespace zmq +{ + + class zmq_engine_t : public io_object_t + { + public: + + zmq_engine_t (class io_thread_t *parent_, object_t *owner_); + + private: + + ~zmq_engine_t (); + + // Handlers for incoming commands. + void process_plug (); + + zmq_engine_t (const zmq_engine_t&); + void operator = (const zmq_engine_t&); + }; + +} + +#endif diff --git a/src/zmq_listener.cpp b/src/zmq_listener.cpp index 1f1e012..9787f7e 100644 --- a/src/zmq_listener.cpp +++ b/src/zmq_listener.cpp @@ -18,9 +18,11 @@ */ #include "zmq_listener.hpp" +#include "zmq_engine.hpp" +#include "io_thread.hpp" #include "err.hpp" -zmq::zmq_listener_t::zmq_listener_t (object_t *parent_, object_t *owner_) : +zmq::zmq_listener_t::zmq_listener_t (io_thread_t *parent_, object_t *owner_) : io_object_t (parent_, owner_) { } @@ -31,5 +33,8 @@ zmq::zmq_listener_t::~zmq_listener_t () void zmq::zmq_listener_t::process_plug () { - // TODO: Register with the I/O thread here. + // TODO: Testing code follows... + object_t *engine = new zmq_engine_t (choose_io_thread (0), owner); + send_plug (engine); + send_own (owner, engine); } diff --git a/src/zmq_listener.hpp b/src/zmq_listener.hpp index 12192b2..ea7cb92 100644 --- a/src/zmq_listener.hpp +++ b/src/zmq_listener.hpp @@ -29,11 +29,12 @@ namespace zmq { public: - zmq_listener_t (object_t *parent_, object_t *owner_); - ~zmq_listener_t (); + zmq_listener_t (class io_thread_t *parent_, object_t *owner_); private: + ~zmq_listener_t (); + // Handlers for incoming commands. void process_plug (); |