Merge pull request #57 from hintjens/master

Backported fixes for LIBZMQ-465

src/decoder.cpp

@@ -57,11 +57,6 @@ void zmq::decoder_t::set_msg_sink (i_msg_sink *msg_sink_)
     msg_sink = msg_sink_;
 }
 
-bool zmq::decoder_t::stalled () const
-{
-    return next == &decoder_t::message_ready;
-}
-
 bool zmq::decoder_t::one_byte_size_ready ()
 {
     //  First byte of size is read. If it is 0xff read 8-byte size.

src/decoder.hpp

@@ -143,6 +143,28 @@ namespace zmq
             }
         }
 
+        //  Returns true if the decoder has been fed all required data
+        //  but cannot proceed with the next decoding step.
+        //  False is returned if the decoder has encountered an error.
+        bool stalled ()
+        {
+            while (!to_read) {
+                if (!(static_cast <T*> (this)->*next) ()) {
+                    if (unlikely (!(static_cast <T*> (this)->next)))
+                        return false;
+                    return true;
+                }
+            }
+
+            return false;
+        }
+
+        inline bool message_ready_size (size_t msg_sz)
+        {
+            zmq_assert (false);
+            return false;
+        }
+
     protected:
 
         //  Prototype of state machine action. Action should return false if
@@ -166,13 +188,13 @@ namespace zmq
             next = NULL;
         }
 
+    private:
+
         //  Next step. If set to NULL, it means that associated data stream
         //  is dead. Note that there can be still data in the process in such
         //  case.
         step_t next;
 
-    private:
-
         //  Where to store the read data.
         unsigned char *read_pos;
 
@@ -199,10 +221,6 @@ namespace zmq
         //  Set the receiver of decoded messages.
         void set_msg_sink (i_msg_sink *msg_sink_);
 
-        //  Returns true if there is a decoded message
-        //  waiting to be delivered to the session.
-        bool stalled () const;
-
     private:
 
         bool one_byte_size_ready ();

src/epoll.cpp

@@ -140,9 +140,10 @@ void zmq::epoll_t::loop ()
         //  Wait for events.
         int n = epoll_wait (epoll_fd, &ev_buf [0], max_io_events,
             timeout ? timeout : -1);
-        if (n == -1 && errno == EINTR)
+        if (n == -1) {
+            errno_assert (errno == EINTR);
             continue;
-        errno_assert (n != -1);
+        }
 
         for (int i = 0; i < n; i ++) {
             poll_entry_t *pe = ((poll_entry_t*) ev_buf [i].data.ptr);

src/i_decoder.hpp

@@ -40,7 +40,7 @@ namespace zmq
 
         virtual size_t process_buffer (unsigned char *data_, size_t size_) = 0;
 
-        virtual bool stalled () const = 0;
+        virtual bool stalled () = 0;
 
     };
 

src/kqueue.cpp

@@ -163,9 +163,10 @@ void zmq::kqueue_t::loop ()
         timespec ts = {timeout / 1000, (timeout % 1000) * 1000000};
         int n = kevent (kqueue_fd, NULL, 0, &ev_buf [0], max_io_events,
             timeout ? &ts: NULL);
-        if (n == -1 && errno == EINTR)
+        if (n == -1) {
+            errno_assert (errno == EINTR);
             continue;
-        errno_assert (n != -1);
+        }
 
         for (int i = 0; i < n; i ++) {
             poll_entry_t *pe = (poll_entry_t*) ev_buf [i].udata;

src/poll.cpp

@@ -125,10 +125,10 @@ void zmq::poll_t::loop ()
 
         //  Wait for events.
         int rc = poll (&pollset [0], pollset.size (), timeout ? timeout : -1);
-        if (rc == -1 && errno == EINTR)
+        if (rc == -1) {
+            errno_assert (errno == EINTR);
             continue;
-        errno_assert (rc != -1);
+        }
-
 
         //  If there are no events (i.e. it's a timeout) there's no point
         //  in checking the pollset.

src/select.cpp

@@ -168,9 +168,10 @@ void zmq::select_t::loop ()
 #else
         int rc = select (maxfd + 1, &readfds, &writefds, &exceptfds,
             timeout ? &tv : NULL);
-        if (rc == -1 && errno == EINTR)
+        if (rc == -1) {
+            errno_assert (errno == EINTR);
             continue;
-        errno_assert (rc != -1);
+        }
 #endif
 
         //  If there are no events (i.e. it's a timeout) there's no point

src/stream_engine.cpp

@@ -50,10 +50,10 @@
 
 zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, const std::string &endpoint_) :
     s (fd_),
+    io_enabled (false),
     inpos (NULL),
     insize (0),
     decoder (NULL),
-    input_error (false),
     outpos (NULL),
     outsize (0),
     encoder (NULL),
@@ -132,6 +132,7 @@ void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
     //  Connect to I/O threads poller object.
     io_object_t::plug (io_thread_);
     handle = add_fd (s);
+    io_enabled = true;
 
     //  Send the 'length' and 'flags' fields of the identity message.
     //  The 'length' field is encoded in the long format.
@@ -153,7 +154,10 @@ void zmq::stream_engine_t::unplug ()
     plugged = false;
 
     //  Cancel all fd subscriptions.
-    rm_fd (handle);
+    if (io_enabled) {
+        rm_fd (handle);
+        io_enabled = false;
+    }
 
     //  Disconnect from I/O threads poller object.
     io_object_t::unplug ();
@@ -225,9 +229,10 @@ void zmq::stream_engine_t::in_event ()
     //  waiting for input events and postpone the termination
     //  until after the session has accepted the message.
     if (disconnection) {
-        input_error = true;
+        if (decoder->stalled ()) {
-        if (decoder->stalled ())
+            rm_fd (handle);
-            reset_pollin (handle);
+            io_enabled = false;
+        }
         else
             error ();
     }
@@ -294,7 +299,7 @@ void zmq::stream_engine_t::activate_out ()
 
 void zmq::stream_engine_t::activate_in ()
 {
-    if (input_error) {
+    if (unlikely (!io_enabled)) {
         //  There was an input error but the engine could not
         //  be terminated (due to the stalled decoder).
         //  Flush the pending message and terminate the engine now.

src/stream_engine.hpp

@@ -96,12 +96,14 @@ namespace zmq
         //  Preamble (10 bytes) + version (1 byte) + socket type (1 byte).
         const static size_t greeting_size = 12;
 
+        //  True iff we are registered with an I/O poller.
+        bool io_enabled;
+
         handle_t handle;
 
         unsigned char *inpos;
         size_t insize;
         i_decoder *decoder;
-        bool input_error;
 
         unsigned char *outpos;
         size_t outsize;

src/v1_decoder.cpp

@@ -58,11 +58,6 @@ void zmq::v1_decoder_t::set_msg_sink (i_msg_sink *msg_sink_)
     msg_sink = msg_sink_;
 }
 
-bool zmq::v1_decoder_t::stalled () const
-{
-    return next == &v1_decoder_t::message_ready;
-}
-
 bool zmq::v1_decoder_t::flags_ready ()
 {
     msg_flags = 0;

src/v1_decoder.hpp

@@ -44,8 +44,6 @@ namespace zmq
         //  i_decoder interface.
         virtual void set_msg_sink (i_msg_sink *msg_sink_);
 
-        virtual bool stalled () const;
-
     private:
 
         bool flags_ready ();