summaryrefslogtreecommitdiff
path: root/src/pipe.cpp
diff options
context:
space:
mode:
authorMartin Lucina <martin@lucina.net>2012-01-23 08:53:25 +0100
committerMartin Lucina <martin@lucina.net>2012-01-23 08:53:25 +0100
commit5ba1cb20fe6f6699cef1cc726718e760cd4c9af1 (patch)
treedf7b144c5325fd8b3c88c49b456fafc24249abe6 /src/pipe.cpp
parenta15854bd92db69fcd0b4444fe1b8fe3610a7acf6 (diff)
Imported Upstream version 2.0.9.dfsgupstream/2.0.9.dfsg
Diffstat (limited to 'src/pipe.cpp')
-rw-r--r--src/pipe.cpp146
1 files changed, 114 insertions, 32 deletions
diff --git a/src/pipe.cpp b/src/pipe.cpp
index 1df64e9..200beb0 100644
--- a/src/pipe.cpp
+++ b/src/pipe.cpp
@@ -21,20 +21,14 @@
#include "pipe.hpp"
-zmq::reader_t::reader_t (object_t *parent_,
- uint64_t hwm_, uint64_t lwm_) :
+zmq::reader_t::reader_t (object_t *parent_, uint64_t lwm_) :
object_t (parent_),
pipe (NULL),
peer (NULL),
- hwm (hwm_),
lwm (lwm_),
msgs_read (0),
endpoint (NULL)
-{
- // Adjust lwm and hwm.
- if (lwm == 0 || lwm > hwm)
- lwm = hwm;
-}
+{}
zmq::reader_t::~reader_t ()
{
@@ -50,15 +44,32 @@ void zmq::reader_t::set_pipe (pipe_t *pipe_)
register_pipe (pipe);
}
+bool zmq::reader_t::is_delimiter (zmq_msg_t &msg_)
+{
+ unsigned char *offset = 0;
+
+ return msg_.content == (void*) (offset + ZMQ_DELIMITER);
+}
+
bool zmq::reader_t::check_read ()
{
// Check if there's an item in the pipe.
- if (pipe->check_read ())
- return true;
-
// If not, deactivate the pipe.
- endpoint->kill (this);
- return false;
+ if (!pipe->check_read ()) {
+ endpoint->kill (this);
+ return false;
+ }
+
+ // If the next item in the pipe is message delimiter,
+ // initiate its termination.
+ if (pipe->probe (is_delimiter)) {
+ if (endpoint)
+ endpoint->detach_inpipe (this);
+ term ();
+ return false;
+ }
+
+ return true;
}
bool zmq::reader_t::read (zmq_msg_t *msg_)
@@ -113,20 +124,28 @@ void zmq::reader_t::process_pipe_term_ack ()
}
zmq::writer_t::writer_t (object_t *parent_,
- uint64_t hwm_, uint64_t lwm_) :
+ uint64_t hwm_, int64_t swap_size_) :
object_t (parent_),
pipe (NULL),
peer (NULL),
hwm (hwm_),
- lwm (lwm_),
msgs_read (0),
msgs_written (0),
+ msg_store (NULL),
+ extra_msg_flag (false),
stalled (false),
+ pending_close (false),
endpoint (NULL)
{
- // Adjust lwm and hwm.
- if (lwm == 0 || lwm > hwm)
- lwm = hwm;
+ if (swap_size_ > 0) {
+ msg_store = new (std::nothrow) msg_store_t (swap_size_);
+ if (msg_store != NULL) {
+ if (msg_store->init () < 0) {
+ delete msg_store;
+ msg_store = NULL;
+ }
+ }
+ }
}
void zmq::writer_t::set_endpoint (i_endpoint *endpoint_)
@@ -136,6 +155,10 @@ void zmq::writer_t::set_endpoint (i_endpoint *endpoint_)
zmq::writer_t::~writer_t ()
{
+ if (extra_msg_flag)
+ zmq_msg_close (&extra_msg);
+
+ delete msg_store;
}
void zmq::writer_t::set_pipe (pipe_t *pipe_)
@@ -147,7 +170,7 @@ void zmq::writer_t::set_pipe (pipe_t *pipe_)
bool zmq::writer_t::check_write ()
{
- if (pipe_full ()) {
+ if (pipe_full () && (msg_store == NULL || msg_store->full () || extra_msg_flag)) {
stalled = true;
return false;
}
@@ -157,29 +180,45 @@ bool zmq::writer_t::check_write ()
bool zmq::writer_t::write (zmq_msg_t *msg_)
{
- if (pipe_full ()) {
- stalled = true;
+ if (!check_write ())
return false;
+
+ if (pipe_full ()) {
+ if (msg_store->store (msg_)) {
+ if (!(msg_->flags & ZMQ_MSG_MORE))
+ msg_store->commit ();
+ } else {
+ extra_msg = *msg_;
+ extra_msg_flag = true;
+ }
+ }
+ else {
+ pipe->write (*msg_, msg_->flags & ZMQ_MSG_MORE);
+ if (!(msg_->flags & ZMQ_MSG_MORE))
+ msgs_written++;
}
- pipe->write (*msg_, msg_->flags & ZMQ_MSG_MORE);
- if (!(msg_->flags & ZMQ_MSG_MORE))
- msgs_written++;
return true;
}
void zmq::writer_t::rollback ()
{
- zmq_msg_t msg;
+ if (extra_msg_flag && extra_msg.flags & ZMQ_MSG_MORE) {
+ zmq_msg_close (&extra_msg);
+ extra_msg_flag = false;
+ }
+ if (msg_store != NULL)
+ msg_store->rollback ();
+
+ zmq_msg_t msg;
// Remove all incomplete messages from the pipe.
while (pipe->unwrite (&msg)) {
zmq_assert (msg.flags & ZMQ_MSG_MORE);
zmq_msg_close (&msg);
- msgs_written--;
}
- if (stalled && endpoint != NULL && !pipe_full()) {
+ if (stalled && endpoint != NULL && check_write ()) {
stalled = false;
endpoint->revive (this);
}
@@ -198,6 +237,14 @@ void zmq::writer_t::term ()
// Rollback any unfinished messages.
rollback ();
+ if (msg_store == NULL || (msg_store->empty () && !extra_msg_flag))
+ write_delimiter ();
+ else
+ pending_close = true;
+}
+
+void zmq::writer_t::write_delimiter ()
+{
// Push delimiter into the pipe.
// Trick the compiler to belive that the tag is a valid pointer.
zmq_msg_t msg;
@@ -205,12 +252,47 @@ void zmq::writer_t::term ()
msg.content = (void*) (offset + ZMQ_DELIMITER);
msg.flags = 0;
pipe->write (msg, false);
- pipe->flush ();
+ flush ();
}
void zmq::writer_t::process_reader_info (uint64_t msgs_read_)
{
+ zmq_msg_t msg;
+
msgs_read = msgs_read_;
+ if (msg_store) {
+
+ // Move messages from backing store into pipe.
+ while (!pipe_full () && !msg_store->empty ()) {
+ msg_store->fetch(&msg);
+ // Write message into the pipe.
+ pipe->write (msg, msg.flags & ZMQ_MSG_MORE);
+ if (!(msg.flags & ZMQ_MSG_MORE))
+ msgs_written++;
+ }
+
+ if (extra_msg_flag) {
+ if (!pipe_full ()) {
+ pipe->write (extra_msg, extra_msg.flags & ZMQ_MSG_MORE);
+ if (!(extra_msg.flags & ZMQ_MSG_MORE))
+ msgs_written++;
+ extra_msg_flag = false;
+ }
+ else if (msg_store->store (&extra_msg)) {
+ if (!(extra_msg.flags & ZMQ_MSG_MORE))
+ msg_store->commit ();
+ extra_msg_flag = false;
+ }
+ }
+
+ if (pending_close && msg_store->empty () && !extra_msg_flag) {
+ write_delimiter ();
+ pending_close = false;
+ }
+
+ flush ();
+ }
+
if (stalled && endpoint != NULL) {
stalled = false;
endpoint->revive (this);
@@ -233,9 +315,9 @@ bool zmq::writer_t::pipe_full ()
}
zmq::pipe_t::pipe_t (object_t *reader_parent_, object_t *writer_parent_,
- uint64_t hwm_) :
- reader (reader_parent_, hwm_, compute_lwm (hwm_)),
- writer (writer_parent_, hwm_, compute_lwm (hwm_))
+ uint64_t hwm_, int64_t swap_size_) :
+ reader (reader_parent_, compute_lwm (hwm_)),
+ writer (writer_parent_, hwm_, swap_size_)
{
reader.set_pipe (this);
writer.set_pipe (this);
@@ -276,6 +358,6 @@ uint64_t zmq::pipe_t::compute_lwm (uint64_t hwm_)
if (hwm_ > max_wm_delta * 2)
return hwm_ - max_wm_delta;
else
- return hwm_ / 2;
+ return (hwm_ + 1) / 2;
}