Problem:

When a ZMQ_STREAM socket reaches its SNDHWM, it becomes impossible to
disconnect a peer. Disconnecting requires sending the routing ID
followed by a 0-byte frame. Currently, xsend returns EAGAIN on the
first frame (the ID) if the pipe is full, preventing the second
disconnect frame from ever being processed.

Solution:
Modified xsend in stream.cpp to allow the routing ID frame to pass
even when the pipe is full. If the subsequent payload frame is
0-bytes, the connection is terminated immediately via terminate().

To prevent state-machine desync, _more_out is reset to false if a
data-bearing payload frame is sent on a full pipe, forcing an EAGAIN
and requiring a clean retry from the user.

Author: guyco3

Makefile.am

@@ -460,7 +460,6 @@ test_apps = \
 	tests/test_probe_router \
 	tests/test_stream \
 	tests/test_stream_empty \
-	tests/test_stream_disconnect_peer \
 	tests/test_stream_disconnect \
 	tests/test_stream_timeout \
 	tests/test_disconnect_inproc \
@@ -636,10 +635,6 @@ tests_test_stream_timeout_SOURCES = tests/test_stream_timeout.cpp
 tests_test_stream_timeout_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
 tests_test_stream_timeout_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
 
-tests_test_stream_disconnect_peer_SOURCES = tests/test_stream_disconnect_peer.cpp
-tests_test_stream_disconnect_peer_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
-tests_test_stream_disconnect_peer_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
-
 tests_test_stream_disconnect_SOURCES = tests/test_stream_disconnect.cpp
 tests_test_stream_disconnect_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
 tests_test_stream_disconnect_CPPFLAGS = ${TESTUTIL_CPPFLAGS}

doc/zmq_disconnect_peer.adoc

@@ -16,7 +16,7 @@ to send messages addressed with that 'routing_id' will fail with 'EHOSTUNREACH'
 until a new connection with a different 'routing_id' is established.
 
 This function is supported on socket types that manage per-peer routing ids:
-'ZMQ_SERVER', 'ZMQ_PEER' and `ZMQ_STREAM`. Calling it on other socket types will fail with
+'ZMQ_SERVER' and 'ZMQ_PEER'. Calling it on other socket types will fail with
 'ENOTSUP'.
 
 

src/stream.cpp

@@ -79,9 +79,6 @@ int zmq::stream_t::xsend (msg_t *msg_)
                 _current_out = out_pipe->pipe;
                 if (!_current_out->check_write ()) {
                     out_pipe->active = false;
-                    _current_out = NULL;
-                    errno = EAGAIN;
-                    return -1;
                 }
             } else {
                 errno = EHOSTUNREACH;
@@ -119,6 +116,13 @@ int zmq::stream_t::xsend (msg_t *msg_)
             _current_out = NULL;
             return 0;
         }
+        if (!_current_out->check_write ()) {
+            // Because we set _more_out to false above, the user is forced 
+            // to resend the Identity frame on their next attempt. 
+            _current_out = NULL;
+            errno = EAGAIN;
+            return -1;
+        }
         const bool ok = _current_out->write (msg_);
         if (likely (ok))
             _current_out->flush ();
@@ -239,32 +243,6 @@ bool zmq::stream_t::xhas_out ()
     return true;
 }
 
-int zmq::stream_t::xdisconnect_peer (uint32_t routing_id_)
-{
-    unsigned char buffer[5];
-    buffer[0] = 0;
-    put_uint32 (buffer + 1, routing_id_);
-
-    blob_t routing_id;
-    routing_id.set (buffer, sizeof buffer);
-
-    out_pipe_t *out_pipe = lookup_out_pipe (routing_id);
-    if (!out_pipe) {
-        errno = EHOSTUNREACH;
-        return -1;
-    }
-
-    out_pipe->pipe->terminate (false);
-
-    // if currently writing to this pipe at same time, reset _current_out and _more_out
-    if (out_pipe->pipe == _current_out) {
-        _current_out = NULL;
-        _more_out = false;
-    }
-
-    return 0;
-}
-
 void zmq::stream_t::identify_peer (pipe_t *pipe_, bool locally_initiated_)
 {
     //  Always assign routing id for raw-socket

src/stream.hpp

@@ -29,7 +29,6 @@ class stream_t ZMQ_FINAL : public routing_socket_base_t
     void xread_activated (zmq::pipe_t *pipe_);
     void xpipe_terminated (zmq::pipe_t *pipe_);
     int xsetsockopt (int option_, const void *optval_, size_t optvallen_);
-    int xdisconnect_peer (uint32_t routing_id_) ZMQ_OVERRIDE;
 
   private:
     //  Generate peer's id and update lookup map

tests/CMakeLists.txt

@@ -26,7 +26,6 @@ set(tests
   test_probe_router
   test_stream
   test_stream_empty
-  test_stream_disconnect_peer
   test_stream_disconnect
   test_disconnect_inproc
   test_unbind_wildcard