diff options
| author | Martin Sustrik <sustrik@250bpm.com> | 2011-05-22 17:26:53 +0200 |
|---|---|---|
| committer | Martin Sustrik <sustrik@250bpm.com> | 2011-05-22 17:26:53 +0200 |
| commit | acf0b0e515515e51ad32ba7a2d147ce703579478 (patch) | |
| tree | d2032009cf46c23aa0f677c2216914f718ab968a /src | |
| parent | 9e6b39925603f9e64db08c469bd628d7ef9465de (diff) | |
Introduces bi-directional pipes
So far, there was a pair of unidirectional pipes between a socket
and a session (or an inproc peer). This resulted in complex
problems with half-closed states and tracking which inpipe
corresponds to which outpipe.
This patch doesn't add any functionality in itself, but is
essential for further work on features like subscription
forwarding.
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
Diffstat (limited to 'src')
| -rw-r--r-- | src/array.hpp | 79 | ||||
| -rw-r--r-- | src/command.hpp | 11 | ||||
| -rw-r--r-- | src/connect_session.cpp | 1 | ||||
| -rw-r--r-- | src/dist.cpp | 10 | ||||
| -rw-r--r-- | src/dist.hpp | 13 | ||||
| -rw-r--r-- | src/fq.cpp | 12 | ||||
| -rw-r--r-- | src/fq.hpp | 12 | ||||
| -rw-r--r-- | src/lb.cpp | 8 | ||||
| -rw-r--r-- | src/lb.hpp | 14 | ||||
| -rw-r--r-- | src/object.cpp | 40 | ||||
| -rw-r--r-- | src/object.hpp | 19 | ||||
| -rw-r--r-- | src/options.cpp | 2 | ||||
| -rw-r--r-- | src/options.hpp | 5 | ||||
| -rw-r--r-- | src/own.cpp | 3 | ||||
| -rw-r--r-- | src/pair.cpp | 98 | ||||
| -rw-r--r-- | src/pair.hpp | 24 | ||||
| -rw-r--r-- | src/pipe.cpp | 335 | ||||
| -rw-r--r-- | src/pipe.hpp | 201 | ||||
| -rw-r--r-- | src/pull.cpp | 26 | ||||
| -rw-r--r-- | src/pull.hpp | 13 | ||||
| -rw-r--r-- | src/push.cpp | 26 | ||||
| -rw-r--r-- | src/push.hpp | 13 | ||||
| -rw-r--r-- | src/session.cpp | 167 | ||||
| -rw-r--r-- | src/session.hpp | 29 | ||||
| -rw-r--r-- | src/socket_base.cpp | 73 | ||||
| -rw-r--r-- | src/socket_base.hpp | 12 | ||||
| -rw-r--r-- | src/xpub.cpp | 26 | ||||
| -rw-r--r-- | src/xpub.hpp | 12 | ||||
| -rw-r--r-- | src/xrep.cpp | 90 | ||||
| -rw-r--r-- | src/xrep.hpp | 24 | ||||
| -rw-r--r-- | src/xreq.cpp | 28 | ||||
| -rw-r--r-- | src/xreq.hpp | 13 | ||||
| -rw-r--r-- | src/xsub.cpp | 26 | ||||
| -rw-r--r-- | src/xsub.hpp | 13 |
34 files changed, 709 insertions, 769 deletions
diff --git a/src/array.hpp b/src/array.hpp index 1d18e48..e7b5266 100644 --- a/src/array.hpp +++ b/src/array.hpp @@ -28,14 +28,17 @@ namespace zmq { // Base class for objects stored in the array. Note that each object can - // be stored in at most one array. + // be stored in at most two arrays. This is needed specifically in the + // case where single pipe object is stored both in array of inbound pipes + // and in the array of outbound pipes. class array_item_t { public: inline array_item_t () : - array_index (-1) + array_index1 (-1), + array_index2 (-1) { } @@ -45,19 +48,30 @@ namespace zmq { } - inline void set_array_index (int index_) + inline void set_array_index1 (int index_) { - array_index = index_; + array_index1 = index_; } - inline int get_array_index () + inline int get_array_index1 () { - return array_index; + return array_index1; + } + + inline void set_array_index2 (int index_) + { + array_index2 = index_; + } + + inline int get_array_index2 () + { + return array_index2; } private: - int array_index; + int array_index1; + int array_index2; array_item_t (const array_item_t&); const array_item_t &operator = (const array_item_t&); @@ -65,9 +79,11 @@ namespace zmq // Fast array implementation with O(1) access to item, insertion and // removal. Array stores pointers rather than objects. The objects have - // to be derived from array_item_t class. + // to be derived from array_item_t class, thus they can be stored in + // two arrays. Template parameter N specifies which index in array_item_t + // to use. - template <typename T> class array_t + template <typename T, int N = 1> class array_t { public: @@ -98,28 +114,48 @@ namespace zmq inline void push_back (T *item_) { - if (item_) - item_->set_array_index ((int) items.size ()); + if (item_) { + if (N == 1) + item_->set_array_index1 ((int) items.size ()); + else + item_->set_array_index2 ((int) items.size ()); + } items.push_back (item_); } - inline void erase (T *item_) { - erase (item_->get_array_index ()); + inline void erase (T *item_) + { + if (N == 1) + erase (item_->get_array_index1 ()); + else + erase (item_->get_array_index2 ()); } inline void erase (size_type index_) { - if (items.back ()) - items.back ()->set_array_index ((int) index_); + if (items.back ()) { + if (N == 1) + items.back ()->set_array_index1 ((int) index_); + else + items.back ()->set_array_index2 ((int) index_); + } items [index_] = items.back (); items.pop_back (); } inline void swap (size_type index1_, size_type index2_) { - if (items [index1_]) - items [index1_]->set_array_index ((int) index2_); - if (items [index2_]) - items [index2_]->set_array_index ((int) index1_); + if (N == 1) { + if (items [index1_]) + items [index1_]->set_array_index1 ((int) index2_); + if (items [index2_]) + items [index2_]->set_array_index1 ((int) index1_); + } + else { + if (items [index1_]) + items [index1_]->set_array_index2 ((int) index2_); + if (items [index2_]) + items [index2_]->set_array_index2 ((int) index1_); + } std::swap (items [index1_], items [index2_]); } @@ -130,7 +166,10 @@ namespace zmq inline size_type index (T *item_) { - return (size_type) item_->get_array_index (); + if (N == 1) + return (size_type) item_->get_array_index1 (); + else + return (size_type) item_->get_array_index2 (); } private: diff --git a/src/command.hpp b/src/command.hpp index 35aed0f..ff7b551 100644 --- a/src/command.hpp +++ b/src/command.hpp @@ -40,8 +40,8 @@ namespace zmq own, attach, bind, - activate_reader, - activate_writer, + activate_read, + activate_write, pipe_term, pipe_term_ack, term_req, @@ -79,8 +79,7 @@ namespace zmq // Sent from session to socket to establish pipe(s) between them. // Caller have used inc_seqnum beforehand sending the command. struct { - class reader_t *in_pipe; - class writer_t *out_pipe; + class pipe_t *pipe; unsigned char peer_identity_size; unsigned char *peer_identity; } bind; @@ -88,13 +87,13 @@ namespace zmq // Sent by pipe writer to inform dormant pipe reader that there // are messages in the pipe. struct { - } activate_reader; + } activate_read; // Sent by pipe reader to inform pipe writer about how many // messages it has read so far. struct { uint64_t msgs_read; - } activate_writer; + } activate_write; // Sent by pipe reader to pipe writer to ask it to terminate // its end of the pipe. diff --git a/src/connect_session.cpp b/src/connect_session.cpp index c0951cb..9a29bf1 100644 --- a/src/connect_session.cpp +++ b/src/connect_session.cpp @@ -22,6 +22,7 @@ #include "zmq_connecter.hpp" #include "pgm_sender.hpp" #include "pgm_receiver.hpp" +#include "err.hpp" zmq::connect_session_t::connect_session_t (class io_thread_t *io_thread_, class socket_base_t *socket_, const options_t &options_, diff --git a/src/dist.cpp b/src/dist.cpp index 7c15bfd..3afa196 100644 --- a/src/dist.cpp +++ b/src/dist.cpp @@ -39,10 +39,8 @@ zmq::dist_t::~dist_t () zmq_assert (pipes.empty ()); } -void zmq::dist_t::attach (writer_t *pipe_) +void zmq::dist_t::attach (pipe_t *pipe_) { - pipe_->set_event_sink (this); - // If we are in the middle of sending a message, we'll add new pipe // into the list of eligible pipes. Otherwise we add it to the list // of active pipes. @@ -74,7 +72,7 @@ void zmq::dist_t::terminate () pipes [i]->terminate (); } -void zmq::dist_t::terminated (writer_t *pipe_) +void zmq::dist_t::terminated (pipe_t *pipe_) { // Remove the pipe from the list; adjust number of active and/or // eligible pipes accordingly. @@ -88,7 +86,7 @@ void zmq::dist_t::terminated (writer_t *pipe_) sink->unregister_term_ack (); } -void zmq::dist_t::activated (writer_t *pipe_) +void zmq::dist_t::activated (pipe_t *pipe_) { // Move the pipe from passive to eligible state. pipes.swap (pipes.index (pipe_), eligible); @@ -153,7 +151,7 @@ bool zmq::dist_t::has_out () return true; } -bool zmq::dist_t::write (class writer_t *pipe_, msg_t *msg_) +bool zmq::dist_t::write (pipe_t *pipe_, msg_t *msg_) { if (!pipe_->write (msg_)) { pipes.swap (pipes.index (pipe_), active - 1); diff --git a/src/dist.hpp b/src/dist.hpp index fd522b9..c137332 100644 --- a/src/dist.hpp +++ b/src/dist.hpp @@ -31,33 +31,32 @@ namespace zmq // Class manages a set of outbound pipes. It sends each messages to // each of them. - class dist_t : public i_writer_events + class dist_t { public: dist_t (class own_t *sink_); ~dist_t (); - void attach (writer_t *pipe_); + void attach (class pipe_t *pipe_); void terminate (); int send (class msg_t *msg_, int flags_); bool has_out (); - // i_writer_events interface implementation. - void activated (writer_t *pipe_); - void terminated (writer_t *pipe_); + void activated (class pipe_t *pipe_); + void terminated (class pipe_t *pipe_); private: // Write the message to the pipe. Make the pipe inactive if writing // fails. In such a case false is returned. - bool write (class writer_t *pipe_, class msg_t *msg_); + bool write (class pipe_t *pipe_, class msg_t *msg_); // Put the message to all active pipes. void distribute (class msg_t *msg_, int flags_); // List of outbound pipes. - typedef array_t <class writer_t> pipes_t; + typedef array_t <class pipe_t, 2> pipes_t; pipes_t pipes; // Number of active pipes. All the active pipes are located at the @@ -38,10 +38,8 @@ zmq::fq_t::~fq_t () zmq_assert (pipes.empty ()); } -void zmq::fq_t::attach (reader_t *pipe_) +void zmq::fq_t::attach (pipe_t *pipe_) { - pipe_->set_event_sink (this); - pipes.push_back (pipe_); pipes.swap (active, pipes.size () - 1); active++; @@ -53,7 +51,7 @@ void zmq::fq_t::attach (reader_t *pipe_) } } -void zmq::fq_t::terminated (reader_t *pipe_) +void zmq::fq_t::terminated (pipe_t *pipe_) { // Make sure that we are not closing current pipe while // message is half-read. @@ -72,10 +70,6 @@ void zmq::fq_t::terminated (reader_t *pipe_) sink->unregister_term_ack (); } -void zmq::fq_t::delimited (reader_t *pipe_) -{ -} - void zmq::fq_t::terminate () { zmq_assert (!terminating); @@ -86,7 +80,7 @@ void zmq::fq_t::terminate () pipes [i]->terminate (); } -void zmq::fq_t::activated (reader_t *pipe_) +void zmq::fq_t::activated (pipe_t *pipe_) { // Move the pipe to the list of active pipes. pipes.swap (pipes.index (pipe_), active); @@ -31,28 +31,26 @@ namespace zmq // Class manages a set of inbound pipes. On receive it performs fair // queueing (RFC970) so that senders gone berserk won't cause denial of // service for decent senders. - class fq_t : public i_reader_events + class fq_t { public: fq_t (class own_t *sink_); ~fq_t (); - void attach (reader_t *pipe_); + void attach (pipe_t *pipe_); void terminate (); int recv (msg_t *msg_, int flags_); bool has_in (); - // i_reader_events implementation. - void activated (reader_t *pipe_); - void terminated (reader_t *pipe_); - void delimited (reader_t *pipe_); + void activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); private: // Inbound pipes. - typedef array_t <reader_t> pipes_t; + typedef array_t <pipe_t, 1> pipes_t; pipes_t pipes; // Number of active pipes. All the active pipes are located at the @@ -39,10 +39,8 @@ zmq::lb_t::~lb_t () zmq_assert (pipes.empty ()); } -void zmq::lb_t::attach (writer_t *pipe_) +void zmq::lb_t::attach (pipe_t *pipe_) { - pipe_->set_event_sink (this); - pipes.push_back (pipe_); pipes.swap (active, pipes.size () - 1); active++; @@ -63,7 +61,7 @@ void zmq::lb_t::terminate () pipes [i]->terminate (); } -void zmq::lb_t::terminated (writer_t *pipe_) +void zmq::lb_t::terminated (pipe_t *pipe_) { pipes_t::size_type index = pipes.index (pipe_); @@ -85,7 +83,7 @@ void zmq::lb_t::terminated (writer_t *pipe_) sink->unregister_term_ack (); } -void zmq::lb_t::activated (writer_t *pipe_) +void zmq::lb_t::activated (pipe_t *pipe_) { // Move the pipe to the list of active pipes. pipes.swap (pipes.index (pipe_), active); @@ -27,28 +27,28 @@ namespace zmq { - // Class manages a set of outbound pipes. On send it load balances + // This class manages a set of outbound pipes. On send it load balances // messages fairly among the pipes. - class lb_t : public i_writer_events + + class lb_t { public: lb_t (class own_t *sink_); ~lb_t (); - void attach (writer_t *pipe_); + void attach (pipe_t *pipe_); void terminate (); int send (msg_t *msg_, int flags_); bool has_out (); - // i_writer_events interface implementation. - void activated (writer_t *pipe_); - void terminated (writer_t *pipe_); + void activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); private: // List of outbound pipes. - typedef array_t <class writer_t> pipes_t; + typedef array_t <class pipe_t, 2> pipes_t; pipes_t pipes; // Number of active pipes. All the active pipes are located at the diff --git a/src/object.cpp b/src/object.cpp index e2ca6d6..0a06d5f 100644 --- a/src/object.cpp +++ b/src/object.cpp @@ -59,12 +59,12 @@ void zmq::object_t::process_command (command_t &cmd_) { switch (cmd_.type) { - case command_t::activate_reader: - process_activate_reader (); + case command_t::activate_read: + process_activate_read (); break; - case command_t::activate_writer: - process_activate_writer (cmd_.args.activate_writer.msgs_read); + case command_t::activate_write: + process_activate_write (cmd_.args.activate_write.msgs_read); break; case command_t::stop: @@ -90,8 +90,8 @@ void zmq::object_t::process_command (command_t &cmd_) break; case command_t::bind: - process_bind (cmd_.args.bind.in_pipe, cmd_.args.bind.out_pipe, - cmd_.args.bind.peer_identity ? blob_t (cmd_.args.bind.peer_identity, + process_bind (cmd_.args.bind.pipe, cmd_.args.bind.peer_identity ? + blob_t (cmd_.args.bind.peer_identity, cmd_.args.bind.peer_identity_size) : blob_t ()); process_seqnum (); break; @@ -236,8 +236,8 @@ void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_, send_command (cmd); } -void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_, - writer_t *out_pipe_, const blob_t &peer_identity_, bool inc_seqnum_) +void zmq::object_t::send_bind (own_t *destination_, pipe_t *pipe_, + const blob_t &peer_identity_, bool inc_seqnum_) { if (inc_seqnum_) destination_->inc_seqnum (); @@ -248,8 +248,7 @@ void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_, #endif cmd.destination = destination_; cmd.type = command_t::bind; - cmd.args.bind.in_pipe = in_pipe_; - cmd.args.bind.out_pipe = out_pipe_; + cmd.args.bind.pipe = pipe_; if (peer_identity_.empty ()) { cmd.args.bind.peer_identity_size = 0; cmd.args.bind.peer_identity = NULL; @@ -267,18 +266,18 @@ void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_, send_command (cmd); } -void zmq::object_t::send_activate_reader (reader_t *destination_) +void zmq::object_t::send_activate_read (pipe_t *destination_) { command_t cmd; #if defined ZMQ_MAKE_VALGRIND_HAPPY memset (&cmd, 0, sizeof (cmd)); #endif cmd.destination = destination_; - cmd.type = command_t::activate_reader; + cmd.type = command_t::activate_read; send_command (cmd); } -void zmq::object_t::send_activate_writer (writer_t *destination_, +void zmq::object_t::send_activate_write (pipe_t *destination_, uint64_t msgs_read_) { command_t cmd; @@ -286,12 +285,12 @@ void zmq::object_t::send_activate_writer (writer_t *destination_, memset (&cmd, 0, sizeof (cmd)); #endif cmd.destination = destination_; - cmd.type = command_t::activate_writer; - cmd.args.activate_writer.msgs_read = msgs_read_; + cmd.type = command_t::activate_write; + cmd.args.activate_write.msgs_read = msgs_read_; send_command (cmd); } -void zmq::object_t::send_pipe_term (writer_t *destination_) +void zmq::object_t::send_pipe_term (pipe_t *destination_) { command_t cmd; #if defined ZMQ_MAKE_VALGRIND_HAPPY @@ -302,7 +301,7 @@ void zmq::object_t::send_pipe_term (writer_t *destination_) send_command (cmd); } -void zmq::object_t::send_pipe_term_ack (reader_t *destination_) +void zmq::object_t::send_pipe_term_ack (pipe_t *destination_) { command_t cmd; #if defined ZMQ_MAKE_VALGRIND_HAPPY @@ -404,18 +403,17 @@ void zmq::object_t::process_attach (i_engine *engine_, zmq_assert (false); } -void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_, - const blob_t &peer_identity_) +void zmq::object_t::process_bind (pipe_t *pipe_, const blob_t &peer_identity_) { zmq_assert (false); } -void zmq::object_t::process_activate_reader () +void zmq::object_t::process_activate_read () { zmq_assert (false); } -void zmq::object_t::process_activate_writer (uint64_t msgs_read_) +void zmq::object_t::process_activate_write (uint64_t msgs_read_) { zmq_assert (false); } diff --git a/src/object.hpp b/src/object.hpp index 0f5e61b..0f47670 100644 --- a/src/object.hpp +++ b/src/object.hpp @@ -66,14 +66,13 @@ namespace zmq void send_attach (class session_t *destination_, struct i_engine *engine_, const blob_t &peer_identity_, bool inc_seqnum_ = true); - void send_bind (class own_t *destination_, - class reader_t *in_pipe_, class writer_t *out_pipe_, + void send_bind (class own_t *destination_, class pipe_t *pipe_, const blob_t &peer_identity_, bool inc_seqnum_ = true); - void send_activate_reader (class reader_t *destination_); - void send_activate_writer (class writer_t *destination_, + void send_activate_read (class pipe_t *destination_); + void send_activate_write (class pipe_t *destination_, uint64_t msgs_read_); - void send_pipe_term (class writer_t *destination_); - void send_pipe_term_ack (class reader_t *destination_); + void send_pipe_term (class pipe_t *destination_); + void send_pipe_term_ack (class pipe_t *destination_); void send_term_req (class own_t *destination_, class own_t *object_); void send_term (class own_t *destination_, int linger_); @@ -89,10 +88,10 @@ namespace zmq virtual void process_own (class own_t *object_); virtual void process_attach (struct i_engine *engine_, const blob_t &peer_identity_); - virtual void process_bind (class reader_t *in_pipe_, - class writer_t *out_pipe_, const blob_t &peer_identity_); - virtual void process_activate_reader (); - virtual void process_activate_writer (uint64_t msgs_read_); + virtual void process_bind (class pipe_t *pipe_, + const blob_t &peer_identity_); + virtual void process_activate_read (); + virtual void process_activate_write (uint64_t msgs_read_); virtual void process_pipe_term (); virtual void process_pipe_term_ack (); virtual void process_term_req (class own_t *object_); diff --git a/src/options.cpp b/src/options.cpp index 897e0f5..271ebdb 100644 --- a/src/options.cpp +++ b/src/options.cpp @@ -38,8 +38,6 @@ zmq::options_t::options_t () : reconnect_ivl_max (0), backlog (100), maxmsgsize (-1), - requires_in (false), - requires_out (false), immediate_connect (true) { } diff --git a/src/options.hpp b/src/options.hpp index 53d0197..fd39a74 100644 --- a/src/options.hpp +++ b/src/options.hpp @@ -75,11 +75,6 @@ namespace zmq // Maximal size of message to handle. int64_t maxmsgsize; - // These options are never set by the user directly. Instead they are - // provided by the specific socket type. - bool requires_in; - bool requires_out; - // If true, when connecting, pipes are created immediately without // waiting for the connection to be established. That way the socket // is not aware of the peer's identity, however, it is able to send diff --git a/src/own.cpp b/src/own.cpp index 4cbfdd6..cdf20a4 100644 --- a/src/own.cpp +++ b/src/own.cpp @@ -173,6 +173,7 @@ void zmq::own_t::process_term (int linger_) void zmq::own_t::register_term_acks (int count_) { term_acks += count_; + printf ("reg %d acks (%p, %d)\n", count_, (void*) this, term_acks); } void zmq::own_t::unregister_term_ack () @@ -180,6 +181,8 @@ void zmq::own_t::unregister_term_ack () zmq_assert (term_acks > 0); term_acks--; + printf ("unreg 1 acks (%p, %d)\n", (void*) this, term_acks); + // This may be a last ack we are waiting for before termination... check_term_acks (); } diff --git a/src/pair.cpp b/src/pair.cpp index d877b54..93a4327 100644 --- a/src/pair.cpp +++ b/src/pair.cpp @@ -25,111 +25,72 @@ zmq::pair_t::pair_t (class ctx_t *parent_, uint32_t tid_) : socket_base_t (parent_, tid_), - inpipe (NULL), - outpipe (NULL), - inpipe_alive (false), - outpipe_alive (false), + pipe (NULL), terminating (false) { options.type = ZMQ_PAIR; - options.requires_in = true; - options.requires_out = true; } zmq::pair_t::~pair_t () { - zmq_assert (!inpipe); - zmq_assert (!outpipe); + zmq_assert (!pipe); } -void zmq::pair_t::xattach_pipes (reader_t *inpipe_, writer_t *outpipe_, - const blob_t &peer_identity_) +void zmq::pair_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (!inpipe && !outpipe); + zmq_assert (!pipe); - inpipe = inpipe_; - inpipe_alive = true; - inpipe->set_event_sink (this); - - outpipe = outpipe_; - outpipe_alive = true; - outpipe->set_event_sink (this); + pipe = pipe_; + pipe->set_event_sink (this); if (terminating) { - register_term_acks (2); - inpipe_->terminate (); - outpipe_->terminate (); + register_term_acks (1); + pipe_->terminate (); } } -void zmq::pair_t::terminated (reader_t *pipe_) -{ - zmq_assert (pipe_ == inpipe); - inpipe = NULL; - inpipe_alive = false; - - if (terminating) - unregister_term_ack (); -} - -void zmq::pair_t::terminated (writer_t *pipe_) +void zmq::pair_t::terminated (pipe_t *pipe_) { - zmq_assert (pipe_ == outpipe); - outpipe = NULL; - outpipe_alive = false; + zmq_assert (pipe_ == pipe); + pipe = NULL; if (terminating) unregister_term_ack (); } -void zmq::pair_t::delimited (reader_t *pipe_) -{ -} - void zmq::pair_t::process_term (int linger_) { terminating = true; - if (inpipe) { + if (pipe) { register_term_acks (1); - inpipe->terminate (); - } - - if (outpipe) { - register_term_acks (1); - outpipe->terminate (); + pipe->terminate (); } socket_base_t::process_term (linger_); } -void zmq::pair_t::activated (class reader_t *pipe_) +void zmq::pair_t::read_activated (pipe_t *pipe_) { - zmq_assert (!inpipe_alive); - inpipe_alive = true; + // There's just one pipe. No lists of active and inactive pipes. + // There's nothing to do here. } -void zmq::pair_t::activated (class writer_t *pipe_) +void zmq::pair_t::write_activated (pipe_t *pipe_) { - zmq_assert (!outpipe_alive); - outpipe_alive = true; + // There's just one pipe. No lists of active and inactive pipes. + // There's nothing to do here. } int zmq::pair_t::xsend (msg_t *msg_, int flags_) { - if (outpipe == NULL || !outpipe_alive) { - errno = EAGAIN; - return -1; - } - - if (!outpipe->write (msg_)) { - outpipe_alive = false; + if (!pipe || !pipe->write (msg_)) { errno = EAGAIN; return -1; } if (!(flags_ & ZMQ_SNDMORE)) - outpipe->flush (); + pipe->flush (); // Detach the original message from the data buffer. int rc = msg_->init (); @@ -144,14 +105,12 @@ int zmq::pair_t::xrecv (msg_t *msg_, int flags_) int rc = msg_->close (); errno_assert (rc == 0); - if (!inpipe_alive || !inpipe || !inpipe->read (msg_)) { - - // No message is available. - inpipe_alive = false; + if (!pipe || !pipe->read (msg_)) { // Initialise the output parameter to be a 0-byte message. rc = msg_->init (); errno_assert (rc == 0); + errno = EAGAIN; return -1; } @@ -160,24 +119,23 @@ int zmq::pair_t::xrecv (msg_t *msg_, int flags_) bool zmq::pair_t::xhas_in () { - if (!inpipe || !inpipe_alive) + if (!pipe) return false; - inpipe_alive = inpipe->check_read (); - return inpipe_alive; + return pipe->check_read (); } bool zmq::pair_t::xhas_out () { - if (!outpipe || !outpipe_alive) + if (!pipe) return false; msg_t msg; int rc = msg.init (); errno_assert (rc == 0); - outpipe_alive = outpipe->check_write (&msg); + bool result = pipe->check_write (&msg); rc = msg.close (); errno_assert (rc == 0); - return outpipe_alive; + return result; } diff --git a/src/pair.hpp b/src/pair.hpp index a10e15a..2cb050a 100644 --- a/src/pair.hpp +++ b/src/pair.hpp @@ -29,8 +29,7 @@ namespace zmq class pair_t : public socket_base_t, - public i_reader_events, - public i_writer_events + public i_pipe_events { public: @@ -38,32 +37,23 @@ namespace zmq ~pair_t (); // Overloads of functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int flags_); int xrecv (class msg_t *msg_, int flags_); bool xhas_in (); bool xhas_out (); - // i_reader_events interface implementation. - void activated (class reader_t *pipe_); - void terminated (class reader_t *pipe_); - void delimited (class reader_t *pipe_); - - // i_writer_events interface implementation. - void activated (class writer_t *pipe_); - void terminated (class writer_t *pipe_); + // i_pipe_events interface implementation. + void read_activated (class pipe_t *pipe_); + void write_activated (class pipe_t *pipe_); + void terminated (class pipe_t *pipe_); private: // Hook into termination process. void process_term (int linger_); - class reader_t *inpipe; - class writer_t *outpipe; - - bool inpipe_alive; - bool outpipe_alive; + class pipe_t *pipe; bool terminating; diff --git a/src/pipe.cpp b/src/pipe.cpp index 36dc808..fb03042 100644 --- a/src/pipe.cpp +++ b/src/pipe.cpp @@ -19,100 +19,123 @@ */ #include <new> +#include <stddef.h> #include "pipe.hpp" -#include "likely.hpp" +#include "err.hpp" -zmq::reader_t::reader_t (object_t *parent_, pipe_t *pipe_, int lwm_) : +int zmq::pipepair (class object_t *parents_ [2], class pipe_t* pipes_ [2], + int hwms_ [2], bool delays_ [2]) +{ + // Creates two pipe objects. These objects are connected by two ypipes, + // each to pass messages in one direction. + + pipe_t::upipe_t *upipe1 = new (std::nothrow) pipe_t::upipe_t (); + alloc_assert (upipe1); + pipe_t::upipe_t *upipe2 = new (std::nothrow) pipe_t::upipe_t (); + alloc_assert (upipe2); + + pipes_ [0] = new (std::nothrow) pipe_t (parents_ [0], upipe1, upipe2, + hwms_ [1], hwms_ [0], delays_ [0]); + alloc_assert (pipes_ [0]); + pipes_ [1] = new (std::nothrow) pipe_t (parents_ [1], upipe2, upipe1, + hwms_ [0], hwms_ [1], delays_ [1]); + alloc_assert (pipes_ [1]); + + pipes_ [0]->set_peer (pipes_ [1]); + pipes_ [1]->set_peer (pipes_ [0]); + + return 0; +} + +zmq::pipe_t::pipe_t (object_t *parent_, upipe_t *inpipe_, upipe_t *outpipe_, + int inhwm_, int outhwm_, bool delay_) : object_t (parent_), - active (true), - pipe (pipe_), - writer (NULL), - lwm (lwm_), + inpipe (inpipe_), + outpipe (outpipe_), + in_active (true), + out_active (true), + hwm (outhwm_), + lwm (compute_lwm (inhwm_)), msgs_read (0), + msgs_written (0), + peers_msgs_read (0), + peer (NULL), sink (NULL), - terminating (false) + terminating (false), + term_recvd (false), + delimited (false), + delay (delay_) { - // Note that writer is not set here. Writer will inform reader about its - // address once it is created (via set_writer method). } -void zmq::reader_t::set_writer (writer_t *writer_) +zmq::pipe_t::~pipe_t () { - zmq_assert (!writer); - writer = writer_; } -zmq::reader_t::~reader_t () +void zmq::pipe_t::set_peer (pipe_t *peer_) { - // Pipe as such is owned and deallocated by reader object. - // The point is that reader processes the last step of terminal - // handshaking (term_ack). - zmq_assert (pipe); - - // First delete all the unread messages in the pipe. We have to do it by - // hand because msg_t doesn't have automatic destructor. - msg_t msg; - while (pipe->read (&msg)) { - int rc = msg.close (); - errno_assert (rc == 0); - } - - delete pipe; + // Peer can be set once only. + zmq_assert (!peer); + peer = peer_; } -void zmq::reader_t::set_event_sink (i_reader_events *sink_) +void zmq::pipe_t::set_event_sink (i_pipe_events *sink_) { + // Sink can be set once only. zmq_assert (!sink); sink = sink_; } -bool zmq::reader_t::is_delimiter (msg_t &msg_) -{ - return msg_.is_delimiter (); -} - -bool zmq::reader_t::check_read () +bool zmq::pipe_t::check_read () { - if (!active) + if (unlikely (!in_active)) return false; // Check if there's an item in the pipe. - if (!pipe->check_read ()) { - active = false; + if (!inpipe->check_read ()) { + in_active = false; return false; } // If the next item in the pipe is message delimiter, - // initiate its termination. - if (pipe->probe (is_delimiter)) { + // initiate termination process. + if (inpipe->probe (is_delimiter)) { msg_t msg; - bool ok = pipe->read (&msg); + bool ok = inpipe->read (&msg); zmq_assert (ok); - if (sink) - sink->delimited (this); - terminate (); + delimited = true; + + // If pipe_term was already received but wasn't processed because + // of pending messages, we can ack it now. + if (terminating) + send_pipe_term_ack (peer); + return false; } return true; } -bool zmq::reader_t::read (msg_t *msg_) +bool zmq::pipe_t::read (msg_t *msg_) { - if (!active) + if (unlikely (!in_active)) return false; - if (!pipe->read (msg_)) { - active = false; + if (!inpipe->read (msg_)) { + in_active = false; return false; } // If delimiter was read, start termination process of the pipe. if (msg_->is_delimiter ()) { - if (sink) - sink->delimited (this); - terminate (); + delimited = true; + + // If pipe_term was already received but wasn't processed because + // of pending messages, we can ack it now. + if (terminating) + send_pipe_term_ack (peer); + return false; } @@ -120,175 +143,148 @@ bool zmq::reader_t::read (msg_t *msg_) msgs_read++; if (lwm > 0 && msgs_read % lwm == 0) - send_activate_writer (writer, msgs_read); + send_activate_write (peer, msgs_read); return true; } -void zmq::reader_t::terminate () +bool zmq::pipe_t::check_write (msg_t *msg_) { - // If termination was already started by the peer, do nothing. - if (terminating) - return; + if (unlikely (!out_active)) + return false; - active = false; - terminating = true; - send_pipe_term (writer); -} + bool full = hwm > 0 && msgs_written - peers_msgs_read == uint64_t (hwm); -void zmq::reader_t::process_activate_reader () -{ - // Forward the event to the sink (either socket or session). - active = true; - sink->activated (this); + if (unlikely (full)) { + out_active = false; + return false; + } + + return true; } -void zmq::reader_t::process_pipe_term_ack () +bool zmq::pipe_t::write (msg_t *msg_) { - // At this point writer may already be deallocated. - // For safety's sake drop the reference to it. - writer = NULL; + if (unlikely (!check_write (msg_))) + return false; - // Notify owner about the termination. - zmq_assert (sink); - sink->terminated (this); + outpipe->write (*msg_, msg_->flags () & msg_t::more); + if (!(msg_->flags () & msg_t::more)) + msgs_written++; - // Deallocate resources. - delete this; + return true; } -zmq::writer_t::writer_t (object_t *parent_, pipe_t *pipe_, reader_t *reader_, - int hwm_) : - object_t (parent_), - active (true), - pipe (pipe_), - reader (reader_), - hwm (hwm_), - msgs_read (0), - msgs_written (0), - sink (NULL), - terminating (false) +void zmq::pipe_t::rollback () { - // Inform reader about the writer. - reader->set_writer (this); + // Remove incomplete message from the outbound pipe. + msg_t msg; + while (outpipe->unwrite (&msg)) { + zmq_assert (msg.flags () & msg_t::more); + int rc = msg.close (); + errno_assert (rc == 0); + } } -zmq::writer_t::~writer_t () +void zmq::pipe_t::flush () { + if (!outpipe->flush ()) + send_activate_read (peer); } -void zmq::writer_t::set_event_sink (i_writer_events *sink_) +void zmq::pipe_t::process_activate_read () { - zmq_assert (!sink); - sink = sink_; + if (!in_active && !terminating) { + in_active = true; + sink->read_activated (this); + } } -bool zmq::writer_t::check_write (msg_t *msg_) +void zmq::pipe_t::process_activate_write (uint64_t msgs_read_) { - // We've already checked and there's no space free for the new message. - // There's no point in checking once again. - if (unlikely (!active)) - return false; + // Remember the peers's message sequence number. + peers_msgs_read = msgs_read_; - if (unlikely (pipe_full ())) { - active = false; - return false; + if (!out_active && !terminating) { + out_active = true; + sink->write_activated (this); } - - return true; } -bool zmq::writer_t::write (msg_t *msg_) +void zmq::pipe_t::process_pipe_term () { - if (unlikely (!check_write (msg_))) - return false; - - pipe->write (*msg_, msg_->flags () & msg_t::more); - if (!(msg_->flags () & msg_t::more)) - msgs_written++; - - return true; + term_recvd = true; + + // We can proceed with the termination if one of the following is true: + // 1. User asked this side of pipe to terminate already. + // 2. Waiting for pending messages in not required. + // 3. Delimiter was already received. + if (terminating || !delay || delimited) { + terminating = true; + send_pipe_term_ack (peer); + } } -void zmq::writer_t::rollback () +void zmq::pipe_t::process_pipe_term_ack () { - // Remove incomplete message from the pipe. + // Notify the user that all the references to the pipe should be dropped. + zmq_assert (sink); + sink->terminated (this); + + // If the peer haven't asked for the termination itself, we have to + // ack the ack, so that it can deallocate properly. + if (!term_recvd) + send_pipe_term_ack (peer); + + // We'll deallocate the inbound pipe, the peer will deallocate the outbound + // pipe (which is an inbound pipe from its point of view). + // First, delete all the unread messages in the pipe. We have to do it by + // hand because msg_t doesn't have automatic destructor. Then deallocate + // the ypipe itself. msg_t msg; - while (pipe->unwrite (&msg)) { - zmq_assert (msg.flags () & msg_t::more); - int rc = msg.close (); - errno_assert (rc == 0); + while (inpipe->read (&msg)) { + int rc = msg.close (); + errno_assert (rc == 0); } -} + delete inpipe; -void zmq::writer_t::flush () -{ - if (!pipe->flush ()) - send_activate_reader (reader); + // Deallocate the pipe object + delete this; } -void zmq::writer_t::terminate () +void zmq::pipe_t::terminate () { // Prevent double termination. if (terminating) return; terminating = true; - // Mark the pipe as not available for writing. - active = false; + // Stop inbound and outbound flow of messages. + in_active = false; + out_active = false; - // Rollback any unfinished messages. + // Rollback any unfinished outbound messages. rollback (); - // Push delimiter into the pipe. Trick the compiler to belive that - // the tag is a valid pointer. Note that watermarks are not checked - // thus the delimiter can be written even though the pipe is full. + // Push delimiter into the outbound pipe. Note that watermarks are not + // checked thus the delimiter can be written even though the pipe is full. msg_t msg; msg.init_delimiter (); - pipe->write (msg, false); + outpipe->write (msg, false); flush (); -} - -void zmq::writer_t::process_activate_writer (uint64_t msgs_read_) -{ - // Store the reader's message sequence number. - msgs_read = msgs_read_; - - // If the writer was non-active before, let's make it active - // (available for writing messages to). - if (!active && !terminating) { - active = true; - zmq_assert (sink); - sink->activated (this); - } -} - -void zmq::writer_t::process_pipe_term () -{ - send_pipe_term_ack (reader); - - // The above command allows reader to deallocate itself and the pipe. - // For safety's sake we'll drop the pointers here. - reader = NULL; - pipe = NULL; - // Notify owner about the termination. - zmq_assert (sink); - sink->terminated (this); - - // Deallocate the resources. - delete this; + // Start the termination handshaking. + send_pipe_term (peer); } -bool zmq::writer_t::pipe_full () +bool zmq::pipe_t::is_delimiter (msg_t &msg_) { - return hwm > 0 && msgs_written - msgs_read == uint64_t (hwm); + return msg_.is_delimiter (); } -void zmq::create_pipe (object_t *reader_parent_, object_t *writer_parent_, - int hwm_, reader_t **reader_, writer_t **writer_) +int zmq::pipe_t::compute_lwm (int hwm_) { - // First compute the low water mark. Following point should be taken + // Compute the low water mark. Following point should be taken // into consideration: // // 1. LWM has to be less than HWM. @@ -308,17 +304,8 @@ void zmq::create_pipe (object_t *reader_parent_, object_t *writer_parent_, // That done, we still we have to account for the cases where // HWM < max_wm_delta thus driving LWM to negative numbers. // Let's make LWM 1/2 of HWM in such cases. - int lwm = (hwm_ > max_wm_delta * 2) ? + int result = (hwm_ > max_wm_delta * 2) ? hwm_ - max_wm_delta : (hwm_ + 1) / 2; - // Create all three objects pipe consists of: the pipe per se, reader and - // writer. The pipe will be handled by reader and writer, its never passed - // to the user. Reader and writer are returned to the user. - pipe_t *pipe = new (std::nothrow) pipe_t (); - alloc_assert (pipe); - *reader_ = new (std::nothrow) reader_t (reader_parent_, pipe, lwm); - alloc_assert (*reader_); - *writer_ = new (std::nothrow) writer_t (writer_parent_, pipe, *reader_, - hwm_); - alloc_assert (*writer_); + return result; } diff --git a/src/pipe.hpp b/src/pipe.hpp index 75b5c47..fcba877 100644 --- a/src/pipe.hpp +++ b/src/pipe.hpp @@ -22,47 +22,43 @@ #define __ZMQ_PIPE_HPP_INCLUDED__ #include "msg.hpp" -#include "array.hpp" #include "ypipe.hpp" #include "config.hpp" #include "object.hpp" -#include "stdint.hpp" +#include "array.hpp" namespace zmq { - // Creates a pipe. Returns pointer to reader and writer objects. - void create_pipe (object_t *reader_parent_, object_t *writer_parent_, - int hwm_, class reader_t **reader_, class writer_t **writer_); - - // The shutdown mechanism for pipe works as follows: Either endpoint - // (or even both of them) can ask pipe to terminate by calling 'terminate' - // method. Pipe then terminates in asynchronous manner. When the part of - // the shutdown tied to the endpoint is done it triggers 'terminated' - // event. When endpoint processes the event and returns, associated - // reader/writer object is deallocated. - - typedef ypipe_t <msg_t, message_pipe_granularity> pipe_t; + // Create a pipepair for bi-directional transfer of messages. + // First HWM is for messages passed from first pipe to the second pipe. + // Second HWM is for messages passed from second pipe to the first pipe. + // Delay specifies whether the pipe receives all the pending messages + // before terminating or whether it terminates straight away. + int pipepair (class object_t *parents_ [2], class pipe_t* pipes_ [2], + int hwms_ [2], bool delays_ [2]); - struct i_reader_events + struct i_pipe_events { - virtual ~i_reader_events () {} + virtual ~i_pipe_events () {} - virtual void terminated (class reader_t *pipe_) = 0; - virtual void activated (class reader_t *pipe_) = 0; - virtual void delimited (class reader_t *pipe_) = 0; + virtual void read_activated (class pipe_t *pipe_) = 0; + virtual void write_activated (class pipe_t *pipe_) = 0; + virtual void terminated (class pipe_t *pipe_) = 0; }; - class reader_t : public object_t, public array_item_t + class pipe_t : + public object_t, + public array_item_t { - friend void create_pipe (object_t*, object_t*, int, - reader_t**, writer_t**); - friend class writer_t; + // This allows pipepair to create pipe objects. + friend int pipepair (class object_t *parents_ [2], + class pipe_t* pipes_ [2], int hwms_ [2], bool delays_ [2]); public: - // Specifies the object to get events from the reader. - void set_event_sink (i_reader_events *endpoint_); + // Specifies the object to send events to. + void set_event_sink (i_pipe_events *sink_); // Returns true if there is at least one message to read in the pipe. bool check_read (); @@ -70,127 +66,100 @@ namespace zmq // Reads a message to the underlying pipe. bool read (msg_t *msg_); - // Ask pipe to terminate. - void terminate (); - - private: - - reader_t (class object_t *parent_, pipe_t *pipe_, int lwm_); - ~reader_t (); - - // To be called only by writer itself! - void set_writer (class writer_t *writer_); - - // Command handlers. - void process_activate_reader (); - void process_pipe_term_ack (); - - // Returns true if the message is delimiter; false otherwise. - static bool is_delimiter (msg_t &msg_); - - // True, if pipe can be read from. - bool active; - - // The underlying pipe. - pipe_t *pipe; - - // Pipe writer associated with the other side of the pipe. - class writer_t *writer; - - // Low watermark for in-memory storage (in bytes). - int lwm; - - // Number of messages read so far. - uint64_t msgs_read; - - // Sink for the events (either the socket of the session). - i_reader_events *sink; - - // True is 'terminate' method was called or delimiter - // was read from the pipe. - bool terminating; - - reader_t (const reader_t&); - const reader_t &operator = (const reader_t&); - }; - - struct i_writer_events - { - virtual ~i_writer_events () {} - - virtual void terminated (class writer_t *pipe_) = 0; - virtual void activated (class writer_t *pipe_) = 0; - }; - - class writer_t : public object_t, public array_item_t - { - friend void create_pipe (object_t*, object_t*, int, - reader_t**, writer_t**); - - public: - - // Specifies the object to get events from the writer. - void set_event_sink (i_writer_events *endpoint_); - - // Checks whether messages can be written to the pipe. - // If writing the message would cause high watermark - // the function returns false. + // Checks whether messages can be written to the pipe. If writing + // the message would cause high watermark the function returns false. bool check_write (msg_t *msg_); // Writes a message to the underlying pipe. Returns false if the // message cannot be written because high watermark was reached. bool write (msg_t *msg_); - // Remove unfinished part of a message from the pipe. + // Remove unfinished parts of the outbound message from the pipe. void rollback (); // Flush the messages downsteam. void flush (); - // Ask pipe to terminate. + // Ask pipe to terminate. The termination will happen asynchronously + // and user will be notified about actual deallocation by 'terminated' + // event. void terminate (); private: - writer_t (class object_t *parent_, pipe_t *pipe_, reader_t *reader_, - int hwm_); - ~writer_t (); - // Command handlers. - void process_activate_writer (uint64_t msgs_read_); + void process_activate_read (); + void process_activate_write (uint64_t msgs_read_); void process_pipe_term (); + void process_pipe_term_ack (); + + // Type of the underlying lock-free pipe. + typedef ypipe_t <msg_t, message_pipe_granularity> upipe_t; - // Tests whether underlying pipe is already full. - bool pipe_full (); + // Constructor is private. Pipe can only be created using + // pipepair function. + pipe_t (object_t *parent_, upipe_t *inpipe_, upipe_t *outpipe_, + int inhwm_, int outhwm_, bool delay_); - // True, if this object can be written to. - bool active; + // Pipepair uses this function to let us know about + // the peer pipe object. + void set_peer (pipe_t *pipe_); - // The underlying pipe. - pipe_t *pipe; + // Destructor is private. Pipe objects destroy themselves. + ~pipe_t (); - // Pipe reader associated with the other side of the pipe. - reader_t *reader; + // Underlying pipes for both directions. + upipe_t *inpipe; + upipe_t *outpipe; - // High watermark for in-memory storage (in bytes). + // Can the pipe be read from / written to? + bool in_active; + bool out_active; + + // High watermark for the outbound pipe. int hwm; - // Last confirmed number of messages read from the pipe. - // The actual number can be higher. - uint64_t msgs_read; + // Low watermark for the inbound pipe. + int lwm; - // Number of messages we have written so far. + // Number of messages read and written so far. + uint64_t msgs_read; uint64_t msgs_written; - // Sink for the events (either the socket or the session). - i_writer_events *sink; + // Last received peer's msgs_read. The actual number in the peer + // can be higher at the moment. + uint64_t peers_msgs_read; - // True is 'terminate' method was called of 'pipe_term' command - // arrived from the reader. + // The pipe object on the other side of the pipepair. + pipe_t *peer; + + // Sink to send events to. + i_pipe_events *sink; + + // True is 'terminate' method was called or termination request + // was received from the peer. bool terminating; - writer_t (const writer_t&); - const writer_t &operator = (const writer_t&); + // True is we've already got pipe_term command from the peer. + bool term_recvd; + + // True if delimiter was already received from the peer. + bool delimited; + + // If true, we receive all the pending inbound messages before + // terminating. If false, we terminate immediately when the peer + // asks us to. + bool delay; + + // Returns true if the message is delimiter; false otherwise. + static bool is_delimiter (msg_t &msg_); + + // Computes appropriate low watermark from the given high watermark. + static int compute_lwm (int hwm_); + + // Disable copying. + pipe_t (const pipe_t&); + const pipe_t &operator = (const pipe_t&); }; } diff --git a/src/pull.cpp b/src/pull.cpp index b9d4433..66457b8 100644 --- a/src/pull.cpp +++ b/src/pull.cpp @@ -27,19 +27,33 @@ zmq::pull_t::pull_t (class ctx_t *parent_, uint32_t tid_) : fq (this) { options.type = ZMQ_PULL; - options.requires_in = true; - options.requires_out = false; } zmq::pull_t::~pull_t () { } -void zmq::pull_t::xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::pull_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (inpipe_ && !outpipe_); - fq.attach (inpipe_); + zmq_assert (pipe_); + pipe_->set_event_sink (this); + fq.attach (pipe_); +} + +void zmq::pull_t::read_activated (pipe_t *pipe_) +{ + fq.activated (pipe_); +} + +void zmq::pull_t::write_activated (pipe_t *pipe_) +{ + // There are no outbound messages in pull socket. This should never happen. + zmq_assert (false); +} + +void zmq::pull_t::terminated (pipe_t *pipe_) +{ + fq.terminated (pipe_); } void zmq::pull_t::process_term (int linger_) diff --git a/src/pull.hpp b/src/pull.hpp index ffc3fdb..af59724 100644 --- a/src/pull.hpp +++ b/src/pull.hpp @@ -22,12 +22,15 @@ #define __ZMQ_PULL_HPP_INCLUDED__ #include "socket_base.hpp" +#include "pipe.hpp" #include "fq.hpp" namespace zmq { - class pull_t : public socket_base_t + class pull_t : + public socket_base_t, + public i_pipe_events { public: @@ -37,13 +40,17 @@ namespace zmq protected: // Overloads of functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xrecv (class msg_t *msg_, int flags_); bool xhas_in (); private: + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); + // Hook into the termination process. void process_term (int linger_); diff --git a/src/push.cpp b/src/push.cpp index d6ee399..12fc8d2 100644 --- a/src/push.cpp +++ b/src/push.cpp @@ -28,19 +28,33 @@ zmq::push_t::push_t (class ctx_t *parent_, uint32_t tid_) : lb (this) { options.type = ZMQ_PUSH; - options.requires_in = false; - options.requires_out = true; } zmq::push_t::~push_t () { } -void zmq::push_t::xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::push_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (!inpipe_ && outpipe_); - lb.attach (outpipe_); + zmq_assert (pipe_); + pipe_->set_event_sink (this); + lb.attach (pipe_); +} + +void zmq::push_t::read_activated (pipe_t *pipe_) +{ + // There are no inbound messages in push socket. This should never happen. + zmq_assert (false); +} + +void zmq::push_t::write_activated (pipe_t *pipe_) +{ + lb.activated (pipe_); +} + +void zmq::push_t::terminated (pipe_t *pipe_) +{ + lb.terminated (pipe_); } void zmq::push_t::process_term (int linger_) diff --git a/src/push.hpp b/src/push.hpp index c4d63f6..67763eb 100644 --- a/src/push.hpp +++ b/src/push.hpp @@ -22,12 +22,15 @@ #define __ZMQ_PUSH_HPP_INCLUDED__ #include "socket_base.hpp" +#include "pipe.hpp" #include "lb.hpp" namespace zmq { - class push_t : public socket_base_t + class push_t : + public socket_base_t, + public i_pipe_events { public: @@ -37,13 +40,17 @@ namespace zmq protected: // Overloads of functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int flags_); bool xhas_out (); private: + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); + // Hook into the termination process. void process_term (int linger_); diff --git a/src/session.cpp b/src/session.cpp index 499fe40..5ef21c7 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -29,13 +29,12 @@ zmq::session_t::session_t (class io_thread_t *io_thread_, class socket_base_t *socket_, const options_t &options_) : own_t (io_thread_, options_), io_object_t (io_thread_), - in_pipe (NULL), + pipe (NULL), incomplete_in (false), - out_pipe (NULL), engine (NULL), socket (socket_), io_thread (io_thread_), - pipes_attached (false), + pipe_attached (false), delimiter_processed (false), force_terminate (false), has_linger_timer (false), @@ -45,8 +44,7 @@ zmq::session_t::session_t (class io_thread_t *io_thread_, zmq::session_t::~session_t () { - zmq_assert (!in_pipe); - zmq_assert (!out_pipe); + zmq_assert (!pipe); if (engine) engine->terminate (); @@ -66,13 +64,9 @@ void zmq::session_t::proceed_with_term () has_linger_timer = false; } - if (in_pipe) { + if (pipe) { register_term_acks (1); - in_pipe->terminate (); - } - if (out_pipe) { - register_term_acks (1); - out_pipe->terminate (); + pipe->terminate (); } // The session has already waited for the linger period. We don't want @@ -82,10 +76,10 @@ void zmq::session_t::proceed_with_term () bool zmq::session_t::read (msg_t *msg_) { - if (!in_pipe) + if (!pipe) return false; - if (!in_pipe->read (msg_)) + if (!pipe->read (msg_)) return false; incomplete_in = msg_->flags () & msg_t::more; @@ -94,7 +88,7 @@ bool zmq::session_t::read (msg_t *msg_) bool zmq::session_t::write (msg_t *msg_) { - if (out_pipe && out_pipe->write (msg_)) { + if (pipe && pipe->write (msg_)) { int rc = msg_->init (); errno_assert (rc == 0); return true; @@ -105,21 +99,20 @@ bool zmq::session_t::write (msg_t *msg_) void zmq::session_t::flush () { - if (out_pipe) - out_pipe->flush (); + if (pipe) + pipe->flush (); } void zmq::session_t::clean_pipes () { - // Get rid of half-processed messages in the out pipe. Flush any - // unflushed messages upstream. - if (out_pipe) { - out_pipe->rollback (); - out_pipe->flush (); - } + if (pipe) { - // Remove any half-read message from the in pipe. - if (in_pipe) { + // Get rid of half-processed messages in the out pipe. Flush any + // unflushed messages upstream. + pipe->rollback (); + pipe->flush (); + + // Remove any half-read message from the in pipe. while (incomplete_in) { msg_t msg; int rc = msg.init (); @@ -134,78 +127,54 @@ void zmq::session_t::clean_pipes () } } -void zmq::session_t::attach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::session_t::attach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (!pipes_attached); - pipes_attached = true; + zmq_assert (!pipe_attached); + pipe_attached = true; - if (inpipe_) { - zmq_assert (!in_pipe); - in_pipe = inpipe_; - in_pipe->set_event_sink (this); - } - - if (outpipe_) { - zmq_assert (!out_pipe); - out_pipe = outpipe_; - out_pipe->set_event_sink (this); + if (pipe_) { + zmq_assert (!pipe); + pipe = pipe_; + pipe->set_event_sink (this); } // If we are already terminating, terminate the pipes straight away. if (state == terminating) { - if (in_pipe) { - in_pipe->terminate (); - register_term_acks (1); - } - if (out_pipe) { - out_pipe->terminate (); + if (pipe) { + pipe->terminate (); register_term_acks (1); } } } -void zmq::session_t::delimited (reader_t *pipe_) +void zmq::session_t::terminated (pipe_t *pipe_) { - zmq_assert (in_pipe == pipe_); - zmq_assert (!delimiter_processed); - delimiter_processed = true; + zmq_assert (pipe == pipe_); - // If we are in process of being closed, but still waiting for all - // pending messeges being sent, we can terminate here. + // If we are in process of being closed, but still waiting for all + // pending messeges being sent, we can terminate here. if (state == pending) proceed_with_term (); -} -void zmq::session_t::terminated (reader_t *pipe_) -{ - zmq_assert (in_pipe == pipe_); - in_pipe = NULL; + pipe = NULL; if (state == terminating) unregister_term_ack (); } -void zmq::session_t::terminated (writer_t *pipe_) +void zmq::session_t::read_activated (pipe_t *pipe_) { - zmq_assert (out_pipe == pipe_); - out_pipe = NULL; - if (state == terminating) - unregister_term_ack (); -} - -void zmq::session_t::activated (reader_t *pipe_) -{ - zmq_assert (in_pipe == pipe_); + zmq_assert (pipe == pipe_); if (likely (engine != NULL)) engine->activate_out (); else - in_pipe->check_read (); + pipe->check_read (); } -void zmq::session_t::activated (writer_t *pipe_) +void zmq::session_t::write_activated (pipe_t *pipe_) { - zmq_assert (out_pipe == pipe_); + zmq_assert (pipe == pipe_); + if (engine) engine->activate_in (); } @@ -240,29 +209,27 @@ void zmq::session_t::process_attach (i_engine *engine_, return; } - // Check whether the required pipes already exist. If not so, we'll - // create them and bind them to the socket object. - if (!pipes_attached) { - zmq_assert (!in_pipe && !out_pipe); - pipes_attached = true; - reader_t *socket_reader = NULL; - writer_t *socket_writer = NULL; - - // Create the pipes, as required. - if (options.requires_in) { - create_pipe (socket, this, options.rcvhwm, &socket_reader, - &out_pipe); - out_pipe->set_event_sink (this); - } - if (options.requires_out) { - create_pipe (this, socket, options.sndhwm, &in_pipe, - &socket_writer); - in_pipe->set_event_sink (this); - } + // Check whether the required pipe already exists and create it + // if it does not. + if (!pipe_attached) { + zmq_assert (!pipe); + pipe_attached = true; + + object_t *parents [2] = {this, socket}; + pipe_t *pipes [2] = {NULL, NULL}; + int hwms [2] = {options.rcvhwm, options.sndhwm}; + bool delays [2] = {true, true}; + int rc = pipepair (parents, pipes, hwms, delays); + errno_assert (rc == 0); + + // Plug the local end of the pipe. + pipes [0]->set_event_sink (this); + + // Remember the local end of the pipe. + pipe = pipes [0]; - // Bind the pipes to the socket object. - if (socket_reader || socket_writer) - send_bind (socket, socket_reader, socket_writer, peer_identity_); + // Ask socket to plug into the remote end of the pipe. + send_bind (socket, pipes [1], peer_identity_); } // Plug in the engine. @@ -282,9 +249,9 @@ void zmq::session_t::detach () // Send the event to the derived class. detached (); - // Just in case, there's only a delimiter in the inbound pipe. - if (in_pipe) - in_pipe->check_read (); + // Just in case there's only a delimiter in the inbound pipe. + if (pipe) + pipe->check_read (); } void zmq::session_t::process_term (int linger_) @@ -308,16 +275,16 @@ void zmq::session_t::process_term (int linger_) // If there's no engine and there's only delimiter in the pipe it wouldn't // be ever read. Thus we check for it explicitly. - if (in_pipe) - in_pipe->check_read (); + if (pipe) + pipe->check_read (); - // If there's no in pipe there are no pending messages to send. + // If there's no in pipe, there are no pending messages to send. // We can proceed with the shutdown straight away. Also, if there is - // inbound pipe, but the delimiter was already processed, we can - // terminate immediately. Alternatively, if the derived session type have + // pipe, but the delimiter was already processed, we can terminate + // immediately. Alternatively, if the derived session type have // called 'terminate' we'll finish straight away. - if (!options.requires_out || delimiter_processed || force_terminate || - (!options.immediate_connect && !in_pipe)) + if (delimiter_processed || force_terminate || + (!options.immediate_connect && !pipe)) proceed_with_term (); } diff --git a/src/session.hpp b/src/session.hpp index d2f8882..4a12d68 100644 --- a/src/session.hpp +++ b/src/session.hpp @@ -34,8 +34,7 @@ namespace zmq public own_t, public io_object_t, public i_inout, - public i_reader_events, - public i_writer_events + public i_pipe_events { public: @@ -50,17 +49,12 @@ namespace zmq void flush (); void detach (); - void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); - - // i_reader_events interface implementation. - void activated (class reader_t *pipe_); - void terminated (class reader_t *pipe_); - void delimited (class reader_t *pipe_); + void attach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); - // i_writer_events interface implementation. - void activated (class writer_t *pipe_); - void terminated (class writer_t *pipe_); + // i_pipe_events interface implementation. + void read_activated (class pipe_t *pipe_); + void write_activated (class pipe_t *pipe_); + void terminated (class pipe_t *pipe_); protected: @@ -103,16 +97,13 @@ namespace zmq // Call this function to move on with the delayed process_term. void proceed_with_term (); - // Inbound pipe, i.e. one the session is getting messages from. - class reader_t *in_pipe; + // Pipe connecting the session to its socket. + class pipe_t *pipe; // This flag is true if the remainder of the message being processed // is still in the in pipe. bool incomplete_in; - // Outbound pipe, i.e. one the socket is sending messages to. - class writer_t *out_pipe; - // The protocol I/O engine connected to the session. struct i_engine *engine; @@ -123,8 +114,8 @@ namespace zmq // the engines into the same thread. class io_thread_t *io_thread; - // If true, pipes were already attached to this session. - bool pipes_attached; + // If true, pipe was already attached to this session. + bool pipe_attached; // If true, delimiter was already read from the inbound pipe. bool delimiter_processed; diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 3e104a8..baa4bd2 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -211,17 +211,17 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_) return 0; } -void zmq::socket_base_t::attach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::socket_base_t::attach_pipe (class pipe_t *pipe_, + const blob_t &peer_identity_) { // If the peer haven't specified it's identity, let's generate one. if (peer_identity_.size ()) { - xattach_pipes (inpipe_, outpipe_, peer_identity_); + xattach_pipe (pipe_, peer_identity_); } else { blob_t identity (17, 0); generate_uuid ((unsigned char*) identity.data () + 1); - xattach_pipes (inpipe_, outpipe_, identity); + xattach_pipe (pipe_, identity); } } @@ -378,11 +378,6 @@ int zmq::socket_base_t::connect (const char *addr_) if (!peer.socket) return -1; - reader_t *inpipe_reader = NULL; - writer_t *inpipe_writer = NULL; - reader_t *outpipe_reader = NULL; - writer_t *outpipe_writer = NULL; - // The total HWM for an inproc connection should be the sum of // the binder's HWM and the connector's HWM. int sndhwm; @@ -396,24 +391,21 @@ int zmq::socket_base_t::connect (const char *addr_) else rcvhwm = options.rcvhwm + peer.options.sndhwm; - // Create inbound pipe, if required. - if (options.requires_in) - create_pipe (this, peer.socket, rcvhwm, &inpipe_reader, - &inpipe_writer); - - // Create outbound pipe, if required. - if (options.requires_out) - create_pipe (peer.socket, this, sndhwm, &outpipe_reader, - &outpipe_writer); + // Create a bi-directional pipe to connect the peers. + object_t *parents [2] = {this, peer.socket}; + pipe_t *pipes [2] = {NULL, NULL}; + int hwms [2] = {sndhwm, rcvhwm}; + bool delays [2] = {true, true}; + int rc = pipepair (parents, pipes, hwms, delays); + errno_assert (rc == 0); - // Attach the pipes to this socket object. - attach_pipes (inpipe_reader, outpipe_writer, peer.options.identity); + // Attach local end of the pipe to this socket object. + attach_pipe (pipes [0], peer.options.identity); - // Attach the pipes to the peer socket. Note that peer's seqnum - // was incremented in find_endpoint function. We don't need it + // Attach remote end of the pipe to the peer socket. Note that peer's + // seqnum was incremented in find_endpoint function. We don't need it // increased here. - send_bind (peer.socket, outpipe_reader, inpipe_writer, - options.identity, false); + send_bind (peer.socket, pipes [1], options.identity, false); return 0; } @@ -435,26 +427,19 @@ int zmq::socket_base_t::connect (const char *addr_) // session once the connection is established. if (options.immediate_connect) { - reader_t *inpipe_reader = NULL; - writer_t *inpipe_writer = NULL; - reader_t *outpipe_reader = NULL; - writer_t *outpipe_writer = NULL; - - // Create inbound pipe, if required. - if (options.requires_in) - create_pipe (this, session, options.rcvhwm, - &inpipe_reader, &inpipe_writer); - - // Create outbound pipe, if required. - if (options.requires_out) - create_pipe (session, this, options.sndhwm, - &outpipe_reader, &outpipe_writer); + // Create a bi-directional pipe. + object_t *parents [2] = {this, session}; + pipe_t *pipes [2] = {NULL, NULL}; + int hwms [2] = {options.sndhwm, options.rcvhwm}; + bool delays [2] = {true, true}; + int rc = pipepair (parents, pipes, hwms, delays); + errno_assert (rc == 0); - // Attach the pipes to the socket object. - attach_pipes (inpipe_reader, outpipe_writer, blob_t ()); + // Attach local end of the pipe to the socket object. + attach_pipe (pipes [0], blob_t ()); - // Attach the pipes to the session object. - session->attach_pipes (outpipe_reader, inpipe_writer, blob_t ()); + // Attach remote end of the pipe to the session object. + session->attach_pipe (pipes [1], blob_t ()); } // Activate the session. Make it a child of this socket. @@ -718,10 +703,10 @@ void zmq::socket_base_t::process_stop () ctx_terminated = true; } -void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_, +void zmq::socket_base_t::process_bind (pipe_t *pipe_, const blob_t &peer_identity_) { - attach_pipes (in_pipe_, out_pipe_, peer_identity_); + attach_pipe (pipe_, peer_identity_); } void zmq::socket_base_t::process_unplug () diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 0a5c574..531751b 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -110,8 +110,8 @@ namespace zmq // Concrete algorithms for the x- methods are to be defined by // individual socket types. - virtual void xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) = 0; + virtual void xattach_pipe (class pipe_t *pipe_, + const blob_t &peer_identity_) = 0; // The default implementation assumes there are no specific socket // options for the particular socket type. If not so, overload this @@ -156,9 +156,8 @@ namespace zmq // bind, is available and compatible with the socket type. int check_protocol (const std::string &protocol_); - // If no identity set generate one and call xattach_pipes (). - void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + // If no identity is set, generate one and call xattach_pipe (). + void attach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); // Processes commands sent to this socket (if any). If 'block' is // set to true, returns only after at least one command was processed. @@ -168,8 +167,7 @@ namespace zmq // Handlers for incoming commands. void process_stop (); - void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_, - const blob_t &peer_identity_); + void process_bind (class pipe_t *pipe_, const blob_t &peer_identity_); void process_unplug (); // Socket's mailbox object. diff --git a/src/xpub.cpp b/src/xpub.cpp index 2b5c4eb..888b42d 100644 --- a/src/xpub.cpp +++ b/src/xpub.cpp @@ -28,19 +28,33 @@ zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_) : dist (this) { options.type = ZMQ_XPUB; - options.requires_in = false; - options.requires_out = true; } zmq::xpub_t::~xpub_t () { } -void zmq::xpub_t::xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (!inpipe_ && outpipe_); - dist.attach (outpipe_); + zmq_assert (pipe_); + pipe_->set_event_sink (this); + dist.attach (pipe_); +} + +void zmq::xpub_t::read_activated (pipe_t *pipe_) +{ + // PUB socket never receives messages. This should never happen. + zmq_assert (false); +} + +void zmq::xpub_t::write_activated (pipe_t *pipe_) +{ + dist.activated (pipe_); +} + +void zmq::xpub_t::terminated (pipe_t *pipe_) +{ + dist.terminated (pipe_); } void zmq::xpub_t::process_term (int linger_) diff --git a/src/xpub.hpp b/src/xpub.hpp index 19aa38a..48efd17 100644 --- a/src/xpub.hpp +++ b/src/xpub.hpp @@ -29,7 +29,9 @@ namespace zmq { - class xpub_t : public socket_base_t + class xpub_t : + public socket_base_t, + public i_pipe_events { public: @@ -37,8 +39,7 @@ namespace zmq ~xpub_t (); // Implementations of virtual functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int flags_); bool xhas_out (); int xrecv (class msg_t *msg_, int flags_); @@ -46,6 +47,11 @@ namespace zmq private: + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); + // Hook into the termination process. void process_term (int linger_); diff --git a/src/xrep.cpp b/src/xrep.cpp index 2650f4e..d82890d 100644 --- a/src/xrep.cpp +++ b/src/xrep.cpp @@ -32,8 +32,6 @@ zmq::xrep_t::xrep_t (class ctx_t *parent_, uint32_t tid_) : terminating (false) { options.type = ZMQ_XREP; - options.requires_in = true; - options.requires_out = true; // On connect, pipes are created only after initial handshaking. // That way we are aware of the peer's identity when binding to the pipes. @@ -46,36 +44,26 @@ zmq::xrep_t::~xrep_t () zmq_assert (outpipes.empty ()); } -void zmq::xrep_t::xattach_pipes (reader_t *inpipe_, writer_t *outpipe_, - const blob_t &peer_identity_) +void zmq::xrep_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - if (outpipe_) { - - outpipe_->set_event_sink (this); - - // TODO: What if new connection has same peer identity as the old one? - outpipe_t outpipe = {outpipe_, true}; - bool ok = outpipes.insert (outpipes_t::value_type ( - peer_identity_, outpipe)).second; - zmq_assert (ok); - - if (terminating) { - register_term_acks (1); - outpipe_->terminate (); - } - } - - if (inpipe_) { - - inpipe_->set_event_sink (this); - - inpipe_t inpipe = {inpipe_, peer_identity_, true}; - inpipes.push_back (inpipe); - - if (terminating) { - register_term_acks (1); - inpipe_->terminate (); - } + zmq_assert (pipe_); + pipe_->set_event_sink (this); + + // Add the pipe to the map out outbound pipes. + // TODO: What if new connection has same peer identity as the old one? + outpipe_t outpipe = {pipe_, true}; + bool ok = outpipes.insert (outpipes_t::value_type ( + peer_identity_, outpipe)).second; + zmq_assert (ok); + + // Add the pipe to the list of inbound pipes. + inpipe_t inpipe = {pipe_, peer_identity_, true}; + inpipes.push_back (inpipe); + + // In case we are already terminating, ask this pipe to terminate as well. + if (terminating) { + register_term_acks (1); + pipe_->terminate (); } } @@ -85,21 +73,17 @@ void zmq::xrep_t::process_term (int linger_) register_term_acks ((int) (inpipes.size () + outpipes.size ())); - for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end (); - ++it) - it->reader->terminate (); - for (outpipes_t::iterator it = outpipes.begin (); it != outpipes.end (); - ++it) - it->second.writer->terminate (); + for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end (); ++it) + it->pipe->terminate (); socket_base_t::process_term (linger_); } -void zmq::xrep_t::terminated (reader_t *pipe_) +void zmq::xrep_t::terminated (pipe_t *pipe_) { for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end (); ++it) { - if (it->reader == pipe_) { + if (it->pipe == pipe_) { if ((inpipes_t::size_type) (it - inpipes.begin ()) < current_in) current_in--; inpipes.erase (it); @@ -107,17 +91,15 @@ void zmq::xrep_t::terminated (reader_t *pipe_) current_in = 0; if (terminating) unregister_term_ack (); - return; + goto clean_outpipes; } } zmq_assert (false); -} -void zmq::xrep_t::terminated (writer_t *pipe_) -{ +clean_outpipes: for (outpipes_t::iterator it = outpipes.begin (); it != outpipes.end (); ++it) { - if (it->second.writer == pipe_) { + if (it->second.pipe == pipe_) { outpipes.erase (it); if (pipe_ == current_out) current_out = NULL; @@ -129,15 +111,11 @@ void zmq::xrep_t::terminated (writer_t *pipe_) zmq_assert (false); } -void zmq::xrep_t::delimited (reader_t *pipe_) -{ -} - -void zmq::xrep_t::activated (reader_t *pipe_) +void zmq::xrep_t::read_activated (pipe_t *pipe_) { for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end (); ++it) { - if (it->reader == pipe_) { + if (it->pipe == pipe_) { zmq_assert (!it->active); it->active = true; return; @@ -146,11 +124,11 @@ void zmq::xrep_t::activated (reader_t *pipe_) zmq_assert (false); } -void zmq::xrep_t::activated (writer_t *pipe_) +void zmq::xrep_t::write_activated (pipe_t *pipe_) { for (outpipes_t::iterator it = outpipes.begin (); it != outpipes.end (); ++it) { - if (it->second.writer == pipe_) { + if (it->second.pipe == pipe_) { zmq_assert (!it->second.active); it->second.active = true; return; @@ -178,7 +156,7 @@ int zmq::xrep_t::xsend (msg_t *msg_, int flags_) outpipes_t::iterator it = outpipes.find (identity); if (it != outpipes.end ()) { - current_out = it->second.writer; + current_out = it->second.pipe; msg_t empty; int rc = empty.init (); errno_assert (rc == 0); @@ -245,7 +223,7 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_) // If we are in the middle of reading a message, just grab next part of it. if (more_in) { zmq_assert (inpipes [current_in].active); - bool fetched = inpipes [current_in].reader->read (msg_); + bool fetched = inpipes [current_in].pipe->read (msg_); zmq_assert (fetched); more_in = msg_->flags () & msg_t::more; if (!more_in) { @@ -261,7 +239,7 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_) // Try to fetch new message. if (inpipes [current_in].active) - prefetched = inpipes [current_in].reader->read (&prefetched_msg); + prefetched = inpipes [current_in].pipe->read (&prefetched_msg); // If we have a message, create a prefix and return it to the caller. if (prefetched) { @@ -311,7 +289,7 @@ bool zmq::xrep_t::xhas_in () // pipe holding messages, skipping only pipes with no messages available. for (inpipes_t::size_type count = inpipes.size (); count != 0; count--) { if (inpipes [current_in].active && - inpipes [current_in].reader->check_read ()) + inpipes [current_in].pipe->check_read ()) return true; // If me don't have a message, mark the pipe as passive and diff --git a/src/xrep.hpp b/src/xrep.hpp index 7ca138c..d0378c2 100644 --- a/src/xrep.hpp +++ b/src/xrep.hpp @@ -35,8 +35,7 @@ namespace zmq // TODO: This class uses O(n) scheduling. Rewrite it to use O(1) algorithm. class xrep_t : public socket_base_t, - public i_reader_events, - public i_writer_events + public i_pipe_events { public: @@ -44,8 +43,7 @@ namespace zmq ~xrep_t (); // Overloads of functions from socket_base_t. - void xattach_pipes (reader_t *inpipe_, writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int flags_); int xrecv (class msg_t *msg_, int flags_); bool xhas_in (); @@ -61,18 +59,14 @@ namespace zmq // Hook into the termination process. void process_term (int linger_); - // i_reader_events interface implementation. - void activated (reader_t *pipe_); - void terminated (reader_t *pipe_); - void delimited (reader_t *pipe_); - - // i_writer_events interface implementation. - void activated (writer_t *pipe_); - void terminated (writer_t *pipe_); + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); struct inpipe_t { - class reader_t *reader; + class pipe_t *pipe; blob_t identity; bool active; }; @@ -95,7 +89,7 @@ namespace zmq struct outpipe_t { - class writer_t *writer; + class pipe_t *pipe; bool active; }; @@ -104,7 +98,7 @@ namespace zmq outpipes_t outpipes; // The pipe we are currently writing to. - class writer_t *current_out; + class pipe_t *current_out; // If true, more outgoing message parts are expected. bool more_out; diff --git a/src/xreq.cpp b/src/xreq.cpp index 2fda2c1..4a6e67e 100644 --- a/src/xreq.cpp +++ b/src/xreq.cpp @@ -28,20 +28,18 @@ zmq::xreq_t::xreq_t (class ctx_t *parent_, uint32_t tid_) : lb (this) { options.type = ZMQ_XREQ; - options.requires_in = true; - options.requires_out = true; } zmq::xreq_t::~xreq_t () { } -void zmq::xreq_t::xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::xreq_t::xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (inpipe_ && outpipe_); - fq.attach (inpipe_); - lb.attach (outpipe_); + zmq_assert (pipe_); + pipe_->set_event_sink (this); + fq.attach (pipe_); + lb.attach (pipe_); } void zmq::xreq_t::process_term (int linger_) @@ -71,3 +69,19 @@ bool zmq::xreq_t::xhas_out () return lb.has_out (); } +void zmq::xreq_t::read_activated (pipe_t *pipe_) +{ + fq.activated (pipe_); +} + +void zmq::xreq_t::write_activated (pipe_t *pipe_) +{ + lb.activated (pipe_); +} + +void zmq::xreq_t::terminated (pipe_t *pipe_) +{ + fq.terminated (pipe_); + lb.terminated (pipe_); +} + diff --git a/src/xreq.hpp b/src/xreq.hpp index e0cafe5..a75e5c8 100644 --- a/src/xreq.hpp +++ b/src/xreq.hpp @@ -23,13 +23,16 @@ #define __ZMQ_XREQ_HPP_INCLUDED__ #include "socket_base.hpp" +#include "pipe.hpp" #include "fq.hpp" #include "lb.hpp" namespace zmq { - class xreq_t : public socket_base_t + class xreq_t : + public socket_base_t, + public i_pipe_events { public: @@ -39,8 +42,7 @@ namespace zmq protected: // Overloads of functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int flags_); int xrecv (class msg_t *msg_, int flags_); bool xhas_in (); @@ -48,6 +50,11 @@ namespace zmq private: + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); + // Hook into the termination process. void process_term (int linger_); diff --git a/src/xsub.cpp b/src/xsub.cpp index b0e8cd2..dc30d71 100644 --- a/src/xsub.cpp +++ b/src/xsub.cpp @@ -30,8 +30,6 @@ zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_) : more (false) { options.type = ZMQ_XSUB; - options.requires_in = true; - options.requires_out = false; int rc = message.init (); errno_assert (rc == 0); } @@ -42,11 +40,27 @@ zmq::xsub_t::~xsub_t () errno_assert (rc == 0); } -void zmq::xsub_t::xattach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) +void zmq::xsub_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) { - zmq_assert (inpipe_ && !outpipe_); - fq.attach (inpipe_); + zmq_assert (pipe_); + pipe_->set_event_sink (this); + fq.attach (pipe_); +} + +void zmq::xsub_t::read_activated (pipe_t *pipe_) +{ + fq.activated (pipe_); +} + +void zmq::xsub_t::write_activated (pipe_t *pipe_) +{ + // SUB socket never sends messages. This should never happen. + zmq_assert (false); +} + +void zmq::xsub_t::terminated (pipe_t *pipe_) +{ + fq.terminated (pipe_); } void zmq::xsub_t::process_term (int linger_) diff --git a/src/xsub.hpp b/src/xsub.hpp index 202a29f..ed9c462 100644 --- a/src/xsub.hpp +++ b/src/xsub.hpp @@ -23,13 +23,16 @@ #include "trie.hpp" #include "socket_base.hpp" +#include "pipe.hpp" #include "msg.hpp" #include "fq.hpp" namespace zmq { - class xsub_t : public socket_base_t + class xsub_t : + public socket_base_t, + public i_pipe_events { public: @@ -39,8 +42,7 @@ namespace zmq protected: // Overloads of functions from socket_base_t. - void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, - const blob_t &peer_identity_); + void xattach_pipe (class pipe_t *pipe_, const blob_t &peer_identity_); int xsend (class msg_t *msg_, int options_); bool xhas_out (); int xrecv (class msg_t *msg_, int flags_); @@ -48,6 +50,11 @@ namespace zmq private: + // i_pipe_events interface implementation. + void read_activated (pipe_t *pipe_); + void write_activated (pipe_t *pipe_); + void terminated (pipe_t *pipe_); + // Hook into the termination process. void process_term (int linger_); |