Imported Debian patch 2.1.3-1debian/2.1.3-1

1 files changed, 236 insertions, 190 deletions

diff --git a/src/pipe.cpp b/src/pipe.cpp
index 200beb0..8dcf577 100644
--- a/src/pipe.cpp
+++ b/src/pipe.cpp
@@ -1,47 +1,72 @@
 /*
-    Copyright (c) 2007-2010 iMatix Corporation
+    Copyright (c) 2007-2011 iMatix Corporation
+    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
 
     This file is part of 0MQ.
 
     0MQ is free software; you can redistribute it and/or modify it under
-    the terms of the Lesser GNU General Public License as published by
+    the terms of the GNU Lesser General Public License as published by
     the Free Software Foundation; either version 3 of the License, or
     (at your option) any later version.
 
     0MQ is distributed in the hope that it will be useful,
     but WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-    Lesser GNU General Public License for more details.
+    GNU Lesser General Public License for more details.
 
-    You should have received a copy of the Lesser GNU General Public License
+    You should have received a copy of the GNU Lesser General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
+#include <new>
+
 #include "../include/zmq.h"
 
 #include "pipe.hpp"
+#include "likely.hpp"
 
-zmq::reader_t::reader_t (object_t *parent_, uint64_t lwm_) :
+zmq::reader_t::reader_t (object_t *parent_, pipe_t *pipe_,
+      uint64_t lwm_) :
     object_t (parent_),
-    pipe (NULL),
-    peer (NULL),
+    active (true),
+    pipe (pipe_),
+    writer (NULL),
     lwm (lwm_),
     msgs_read (0),
-    endpoint (NULL)
-{}
+    sink (NULL),
+    terminating (false)
+{
+    //  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_assert (!writer);
+    writer = writer_;
+}
 
 zmq::reader_t::~reader_t ()
 {
-    if (pipe)
-        unregister_pipe (pipe);
+    //  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 zmq_msg_t is a POD, not a class, so there's no associated
+    //  destructor.
+    zmq_msg_t msg;
+    while (pipe->read (&msg))
+       zmq_msg_close (&msg);
+
+    delete pipe;
 }
 
-void zmq::reader_t::set_pipe (pipe_t *pipe_)
+void zmq::reader_t::set_event_sink (i_reader_events *sink_)
 {
-    zmq_assert (!pipe);
-    pipe = pipe_;
-    peer = &pipe->writer;
-    register_pipe (pipe);
+    zmq_assert (!sink);
+    sink = sink_;
 }
 
 bool zmq::reader_t::is_delimiter (zmq_msg_t &msg_)
@@ -53,19 +78,24 @@ bool zmq::reader_t::is_delimiter (zmq_msg_t &msg_)
 
 bool zmq::reader_t::check_read ()
 {
+    if (!active)
+        return false;
+
     //  Check if there's an item in the pipe.
-    //  If not, deactivate the pipe.
     if (!pipe->check_read ()) {
-        endpoint->kill (this);
+        active = false;
         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 ();
+        zmq_msg_t msg;
+        bool ok = pipe->read (&msg);
+        zmq_assert (ok);
+        if (sink)
+            sink->delimited (this);
+        terminate ();
         return false;
     }
 
@@ -74,17 +104,20 @@ bool zmq::reader_t::check_read ()
 
 bool zmq::reader_t::read (zmq_msg_t *msg_)
 {
+    if (!active)
+        return false;
+
     if (!pipe->read (msg_)) {
-        endpoint->kill (this);
+        active = false;
         return false;
     }
 
     //  If delimiter was read, start termination process of the pipe.
     unsigned char *offset = 0;
     if (msg_->content == (void*) (offset + ZMQ_DELIMITER)) {
-        if (endpoint)
-            endpoint->detach_inpipe (this);
-        term ();
+        if (sink)
+            sink->delimited (this);
+        terminate ();
         return false;
     }
 
@@ -92,87 +125,104 @@ bool zmq::reader_t::read (zmq_msg_t *msg_)
         msgs_read++;
 
     if (lwm > 0 && msgs_read % lwm == 0)
-        send_reader_info (peer, msgs_read);
+        send_activate_writer (writer, msgs_read);
 
     return true;
 }
 
-void zmq::reader_t::set_endpoint (i_endpoint *endpoint_)
+void zmq::reader_t::terminate ()
 {
-    endpoint = endpoint_;
-}
+    //  If termination was already started by the peer, do nothing.
+    if (terminating)
+        return;
 
-void zmq::reader_t::term ()
-{
-    endpoint = NULL;
-    send_pipe_term (peer);
+    active = false;
+    terminating = true;
+    send_pipe_term (writer);
 }
 
-void zmq::reader_t::process_revive ()
+void zmq::reader_t::process_activate_reader ()
 {
-    //  Beacuse of command throttling mechanism, incoming termination request
-    //  may not have been processed before subsequent send.
-    //  In that case endpoint is NULL.
-    if (endpoint)
-        endpoint->revive (this);
+    //  Forward the event to the sink (either socket or session).
+    active = true;
+    sink->activated (this);
 }
 
 void zmq::reader_t::process_pipe_term_ack ()
 {
-    peer = NULL;
-    delete pipe;
+    //  At this point writer may already be deallocated.
+    //  For safety's sake drop the reference to it.
+    writer = NULL;
+
+    //  Notify owner about the termination.
+    zmq_assert (sink);
+    sink->terminated (this);
+
+    //  Deallocate resources.
+    delete this;
 }
 
-zmq::writer_t::writer_t (object_t *parent_,
+zmq::writer_t::writer_t (object_t *parent_, pipe_t *pipe_, reader_t *reader_,
       uint64_t hwm_, int64_t swap_size_) :
     object_t (parent_),
-    pipe (NULL),
-    peer (NULL),
+    active (true),
+    pipe (pipe_),
+    reader (reader_),
     hwm (hwm_),
     msgs_read (0),
     msgs_written (0),
-    msg_store (NULL),
-    extra_msg_flag (false),
-    stalled (false),
-    pending_close (false),
-    endpoint (NULL)
+    swap (NULL),
+    sink (NULL),
+    swapping (false),
+    pending_delimiter (false),
+    terminating (false)
 {
+    //  Inform reader about the writer.
+    reader->set_writer (this);
+
+    //  Open the swap file, if required.
     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;
-            }
-        }
+        swap = new (std::nothrow) swap_t (swap_size_);
+        alloc_assert (swap);
+        int rc = swap->init ();
+        zmq_assert (rc == 0);
     }
 }
 
-void zmq::writer_t::set_endpoint (i_endpoint *endpoint_)
-{
-    endpoint = endpoint_;
-}
-
 zmq::writer_t::~writer_t ()
 {
-    if (extra_msg_flag)
-        zmq_msg_close (&extra_msg);
-
-    delete msg_store;
+    if (swap)
+       delete swap;
 }
 
-void zmq::writer_t::set_pipe (pipe_t *pipe_)
+void zmq::writer_t::set_event_sink (i_writer_events *sink_)
 {
-    zmq_assert (!pipe);
-    pipe = pipe_;
-    peer = &pipe->reader;
+    zmq_assert (!sink);
+    sink = sink_;
 }
 
-bool zmq::writer_t::check_write ()
+bool zmq::writer_t::check_write (zmq_msg_t *msg_)
 {
-    if (pipe_full () && (msg_store == NULL || msg_store->full () || extra_msg_flag)) {
-        stalled = true;
+    //  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;
+
+    if (unlikely (swapping)) {
+        if (unlikely (!swap->fits (msg_))) {
+            active = false;
+            return false;
+        }
+    }
+    else {
+        if (unlikely (pipe_full ())) {
+            if (swap)
+                swapping = true;
+            else {
+                active = false;
+                return false;
+            }
+        }
     }
 
     return true;
@@ -180,73 +230,68 @@ bool zmq::writer_t::check_write ()
 
 bool zmq::writer_t::write (zmq_msg_t *msg_)
 {
-    if (!check_write ())
+    if (unlikely (!check_write (msg_)))
         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 (unlikely (swapping)) {
+        bool stored = swap->store (msg_);
+        zmq_assert (stored);
         if (!(msg_->flags & ZMQ_MSG_MORE))
-            msgs_written++;
+            swap->commit ();
+        return true;
     }
 
+    pipe->write (*msg_, msg_->flags & ZMQ_MSG_MORE);
+    if (!(msg_->flags & ZMQ_MSG_MORE))
+        msgs_written++;
+
     return true;
 }
 
 void zmq::writer_t::rollback ()
 {
-    if (extra_msg_flag && extra_msg.flags & ZMQ_MSG_MORE) {
-        zmq_msg_close (&extra_msg);
-        extra_msg_flag = false;
+    //  Remove incomplete message from the swap.
+    if (unlikely (swapping)) {
+        swap->rollback ();
+        return;
     }
 
-    if (msg_store != NULL)
-        msg_store->rollback ();
-
+    //  Remove incomplete message from the pipe.
     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);
     }
-
-    if (stalled && endpoint != NULL && check_write ()) {
-        stalled = false;
-        endpoint->revive (this);
-    }
 }
 
 void zmq::writer_t::flush ()
 {
-    if (!pipe->flush ())
-        send_revive (peer);
+    //  In the swapping mode, flushing is automatically handled by swap object.
+    if (!swapping && !pipe->flush ())
+        send_activate_reader (reader);
 }
 
-void zmq::writer_t::term ()
+void zmq::writer_t::terminate ()
 {
-    endpoint = NULL;
+    //  Prevent double termination.
+    if (terminating)
+        return;
+    terminating = true;
+
+    //  Mark the pipe as not available for writing.
+    active = false;
 
     //  Rollback any unfinished messages.
     rollback ();
 
-    if (msg_store == NULL || (msg_store->empty () && !extra_msg_flag))
-        write_delimiter ();
-    else
-        pending_close = true;
-}
+    if (swapping) {
+        pending_delimiter = true;
+        return;
+    }
 
-void zmq::writer_t::write_delimiter ()
-{
-    //  Push delimiter into the pipe.
-    //  Trick the compiler to belive that the tag is a valid pointer.
+    //  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.
     zmq_msg_t msg;
     const unsigned char *offset = 0;
     msg.content = (void*) (offset + ZMQ_DELIMITER);
@@ -255,109 +300,110 @@ void zmq::writer_t::write_delimiter ()
     flush ();
 }
 
-void zmq::writer_t::process_reader_info (uint64_t msgs_read_)
+void zmq::writer_t::process_activate_writer (uint64_t msgs_read_)
 {
-    zmq_msg_t msg;
-
+    //  Store the reader's message sequence number.
     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.
+    //  If we are in the swapping mode, we have some messages in the swap.
+    //  Given that pipe is now ready for writing we can move part of the
+    //  swap into the pipe.
+    if (swapping) {
+        zmq_msg_t msg;
+        while (!pipe_full () && !swap->empty ()) {
+            swap->fetch(&msg);
             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 (!pipe->flush ())
+            send_activate_reader (reader);
+
+        //  There are no more messages in the swap. We can switch into
+        //  standard in-memory mode.
+        if (swap->empty ()) {        
+            swapping = false;
+
+            //  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.
+            if (pending_delimiter) {
+                zmq_msg_t msg;
+                const unsigned char *offset = 0;
+                msg.content = (void*) (offset + ZMQ_DELIMITER);
+                msg.flags = 0;
+                pipe->write (msg, false);
+                flush ();
+                return;
             }
         }
-
-        if (pending_close && msg_store->empty () && !extra_msg_flag) {
-            write_delimiter ();
-            pending_close = false;
-        }
-
-        flush ();
     }
 
-    if (stalled && endpoint != NULL) {
-        stalled = false;
-        endpoint->revive (this);
+    //  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 ()
 {
-    if (endpoint)
-        endpoint->detach_outpipe (this);
+    send_pipe_term_ack (reader);
 
-    reader_t *p = peer;
-    peer = NULL;
-    send_pipe_term_ack (p);
-}
+    //  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;
 
-bool zmq::writer_t::pipe_full ()
-{
-    return hwm > 0 && msgs_written - msgs_read == hwm;
-}
+    //  Notify owner about the termination.
+    zmq_assert (sink);
+    sink->terminated (this);
 
-zmq::pipe_t::pipe_t (object_t *reader_parent_, object_t *writer_parent_,
-      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);
+    //  Deallocate the resources.
+    delete this;
 }
 
-zmq::pipe_t::~pipe_t ()
+bool zmq::writer_t::pipe_full ()
 {
-    //  Deallocate all the unread messages in the pipe. We have to do it by
-    //  hand because zmq_msg_t is a POD, not a class, so there's no associated
-    //  destructor.
-    zmq_msg_t msg;
-    while (read (&msg))
-       zmq_msg_close (&msg);
+    return hwm > 0 && msgs_written - msgs_read == hwm;
 }
 
-uint64_t zmq::pipe_t::compute_lwm (uint64_t hwm_)
+void zmq::create_pipe (object_t *reader_parent_, object_t *writer_parent_,
+    uint64_t hwm_, int64_t swap_size_, reader_t **reader_, writer_t **writer_)
 {
-   //  Following point should be taken into consideration when computing
-   //  low watermark:
-   //
-   //  1. LWM has to be less than HWM.
-   //  2. LWM cannot be set to very low value (such as zero) as after filling
-   //     the queue it would start to refill only after all the messages are
-   //     read from it and thus unnecessarily hold the progress back.
-   //  3. LWM cannot be set to very high value (such as HWM-1) as it would
-   //     result in lock-step filling of the queue - if a single message is read
-   //     from a full queue, writer thread is resumed to write exactly one
-   //     message to the queue and go back to sleep immediately. This would
-   //     result in low performance.
-   //
-   //  Given the 3. it would be good to keep HWM and LWM as far apart as
-   //  possible to reduce the thread switching overhead to almost zero,
-   //  say HWM-LWM should be 500 (max_wm_delta).
-   //
-   //  That done, we still we have to account for the cases where HWM<500 thus
-   //  driving LWM to negative numbers. Let's make LWM 1/2 of HWM in such cases.
-
-    if (hwm_ > max_wm_delta * 2)
-        return hwm_ - max_wm_delta;
-    else
-        return (hwm_ + 1) / 2;
+    //  First compute the low water mark. Following point should be taken
+    //  into consideration:
+    //
+    //  1. LWM has to be less than HWM.
+    //  2. LWM cannot be set to very low value (such as zero) as after filling
+    //     the queue it would start to refill only after all the messages are
+    //     read from it and thus unnecessarily hold the progress back.
+    //  3. LWM cannot be set to very high value (such as HWM-1) as it would
+    //     result in lock-step filling of the queue - if a single message is
+    //     read from a full queue, writer thread is resumed to write exactly one
+    //     message to the queue and go back to sleep immediately. This would
+    //     result in low performance.
+    //
+    //  Given the 3. it would be good to keep HWM and LWM as far apart as
+    //  possible to reduce the thread switching overhead to almost zero,
+    //  say HWM-LWM should be max_wm_delta.
+    //
+    //  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.
+    uint64_t lwm = (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_, swap_size_);
+    alloc_assert (*writer_);
 }
-