summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@fastmq.commkdir>2009-08-09 09:24:48 +0200
committerMartin Sustrik <sustrik@fastmq.commkdir>2009-08-09 09:24:48 +0200
commit5b5b513330e96e3e08d0c2c60d03044091976420 (patch)
tree1eb97606ff6827760c0b063aca95470df6ae20a1
parenta8b410e66c3c75809c8e9c01dd3e35c579f02347 (diff)
socket options interface modeled as in BSD sockets
-rw-r--r--include/zmq.h28
-rw-r--r--include/zmq.hpp14
-rw-r--r--src/Makefile.am2
-rw-r--r--src/epoll.cpp4
-rw-r--r--src/i_api.hpp6
-rw-r--r--src/io_object.cpp45
-rw-r--r--src/io_object.hpp23
-rw-r--r--src/io_thread.hpp2
-rw-r--r--src/socket_base.cpp33
-rw-r--r--src/socket_base.hpp9
-rw-r--r--src/zmq.cpp13
-rw-r--r--src/zmq_engine.cpp35
-rw-r--r--src/zmq_engine.hpp47
-rw-r--r--src/zmq_listener.cpp9
-rw-r--r--src/zmq_listener.hpp5
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 ();