diff options
Diffstat (limited to 'src/pipe.hpp')
| -rw-r--r-- | src/pipe.hpp | 152 |
1 files changed, 91 insertions, 61 deletions
diff --git a/src/pipe.hpp b/src/pipe.hpp index ece678a..8898bbc 100644 --- a/src/pipe.hpp +++ b/src/pipe.hpp @@ -23,25 +23,48 @@ #include "../include/zmq.h" #include "stdint.hpp" -#include "i_endpoint.hpp" -#include "yarray_item.hpp" +#include "array.hpp" #include "ypipe.hpp" -#include "msg_store.hpp" +#include "swap.hpp" #include "config.hpp" #include "object.hpp" namespace zmq { - class reader_t : public object_t, public yarray_item_t + // Creates a pipe. Returns pointer to reader and writer objects. + void create_pipe (object_t *reader_parent_, object_t *writer_parent_, + uint64_t hwm_, int64_t swap_size_, 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 <zmq_msg_t, message_pipe_granularity> pipe_t; + + struct i_reader_events { - public: + virtual ~i_reader_events () {}; - reader_t (class object_t *parent_, uint64_t lwm_); - ~reader_t (); + virtual void terminated (class reader_t *pipe_) = 0; + virtual void activated (class reader_t *pipe_) = 0; + virtual void delimited (class reader_t *pipe_) = 0; + }; + + class reader_t : public object_t, public array_item_t + { + friend void create_pipe (object_t*, object_t*, uint64_t, + int64_t, reader_t**, writer_t**); + friend class writer_t; + + public: - void set_pipe (class pipe_t *pipe_); - void set_endpoint (i_endpoint *endpoint_); + // Specifies the object to get events from the reader. + void set_event_sink (i_reader_events *endpoint_); // Returns true if there is at least one message to read in the pipe. bool check_read (); @@ -50,22 +73,31 @@ namespace zmq bool read (zmq_msg_t *msg_); // Ask pipe to terminate. - void term (); + void terminate (); private: + reader_t (class object_t *parent_, pipe_t *pipe_, uint64_t lwm_); + ~reader_t (); + + // To be called only by writer itself! + void set_writer (class writer_t *writer_); + // Command handlers. - void process_revive (); + void process_activate_reader (); void process_pipe_term_ack (); // Returns true if the message is delimiter; false otherwise. static bool is_delimiter (zmq_msg_t &msg_); + // True, if pipe can be read from. + bool active; + // The underlying pipe. - class pipe_t *pipe; + pipe_t *pipe; // Pipe writer associated with the other side of the pipe. - class writer_t *peer; + class writer_t *writer; // Low watermark for in-memory storage (in bytes). uint64_t lwm; @@ -73,26 +105,38 @@ namespace zmq // Number of messages read so far. uint64_t msgs_read; - // Endpoint (either session or socket) the pipe is attached to. - i_endpoint *endpoint; + // 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&); void operator = (const reader_t&); }; - class writer_t : public object_t, public yarray_item_t + struct i_writer_events { - public: + virtual ~i_writer_events () {}; - writer_t (class object_t *parent_, uint64_t hwm_, int64_t swap_size_); - ~writer_t (); + virtual void terminated (class writer_t *pipe_) = 0; + virtual void activated (class writer_t *pipe_) = 0; + }; - void set_pipe (class pipe_t *pipe_); - void set_endpoint (i_endpoint *endpoint_); + class writer_t : public object_t, public array_item_t + { + friend void create_pipe (object_t*, object_t*, uint64_t, + int64_t, reader_t**, writer_t**); + + public: + + // Specifies the object to get events from the writer. + void set_event_sink (i_writer_events *endpoint_); // Checks whether a message can be written to the pipe. - // If writing the message would cause high watermark to be - // exceeded, the function returns false. + // If writing the message would cause high watermark and (optionally) + // swap to be exceeded, the function returns false. bool check_write (); // Writes a message to the underlying pipe. Returns false if the @@ -106,27 +150,31 @@ namespace zmq void flush (); // Ask pipe to terminate. - void term (); + void terminate (); private: - void process_reader_info (uint64_t msgs_read_); + writer_t (class object_t *parent_, pipe_t *pipe_, reader_t *reader_, + uint64_t hwm_, int64_t swap_size_); + ~writer_t (); // Command handlers. + void process_activate_writer (uint64_t msgs_read_); void process_pipe_term (); - // Tests whether the pipe is already full. + // Tests whether underlying pipe is already full. The swap is not + // taken into account. bool pipe_full (); - // Write special message to the pipe so that the reader - // can find out we are finished. - void write_delimiter (); + // True, if this object can be written to. Undelying ypipe may be full + // but as long as there's swap space available, this flag is true. + bool active; // The underlying pipe. - class pipe_t *pipe; + pipe_t *pipe; // Pipe reader associated with the other side of the pipe. - class reader_t *peer; + reader_t *reader; // High watermark for in-memory storage (in bytes). uint64_t hwm; @@ -138,46 +186,28 @@ namespace zmq // Number of messages we have written so far. uint64_t msgs_written; - // Pointer to backing store. If NULL, messages are always + // Pointer to the message swap. If NULL, messages are always // kept in main memory. - msg_store_t *msg_store; - - bool extra_msg_flag; + swap_t *swap; - zmq_msg_t extra_msg; + // Sink for the events (either the socket or the session). + i_writer_events *sink; - // True iff the last attempt to write a message has failed. - bool stalled; + // If true, swap is active. New messages are to be written to the swap. + bool swapping; - bool pending_close; + // If true, there's a delimiter to be written to the pipe after the + // swap is empied. + bool pending_delimiter; - // Endpoint (either session or socket) the pipe is attached to. - i_endpoint *endpoint; + // True is 'terminate' method was called of 'pipe_term' command + // arrived from the reader. + bool terminating; writer_t (const writer_t&); void operator = (const writer_t&); }; - // Message pipe. - class pipe_t : public ypipe_t <zmq_msg_t, message_pipe_granularity> - { - public: - - pipe_t (object_t *reader_parent_, object_t *writer_parent_, - uint64_t hwm_, int64_t swap_size_); - ~pipe_t (); - - reader_t reader; - writer_t writer; - - private: - - uint64_t compute_lwm (uint64_t hwm_); - - pipe_t (const pipe_t&); - void operator = (const pipe_t&); - }; - } #endif |