summaryrefslogtreecommitdiff
path: root/src/pipe.hpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/pipe.hpp')
-rw-r--r--src/pipe.hpp152
1 files changed, 91 insertions, 61 deletions
diff --git a/src/pipe.hpp b/src/pipe.hpp
index ece678a..b154b12 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