Merge pull request zeromq#25 from hintjens/master

Merged fix for zeromq#1382

.gitignore

@@ -74,6 +74,7 @@ test_linger
 test_security_null
 test_security_plain
 test_proxy
+test_proxy_terminate
 test_abstract_ipc
 test_filter_ipc
 test_connect_delay_tipc

AUTHORS

@@ -86,6 +86,7 @@ Philip Kovacs
 Pieter Hintjens 
 Piotr Trojanek 
 Richard Newton 
+Rik van der Heijden
 Robert G. Jakabosky 
 Sebastian Otaegui 
 Stefan Radomski 

Makefile.am

@@ -320,6 +320,7 @@ test_apps = \
  test_inproc_connect \
  test_issue_566 \
  test_proxy \
+ test_proxy_terminate \
  test_many_sockets \
  test_ipc_wildcard \
  test_diffserv \

NEWS

@@ -1,4 +1,4 @@
-0MQ version 4.1.1 rc2, released on 2014/12/xx
+0MQ version 4.1.1 rc2, released on 2015/xx/xx
 =============================================
 
 * Fixed #1208 - fix recursion in automake packaging
@@ -17,6 +17,8 @@
 
 * Fixed #1389 - PUB, PUSH sockets had slow memory leak.
 
+* Fixed #1382 - zmq_proxy did not terminate if there were no readers.
+
 
 0MQ version 4.1.0 rc1, released on 2014/10/14
 =============================================

src/proxy.cpp

@@ -159,14 +159,16 @@ int zmq::proxy (
  }
  // Process a request
  if (state == active
- && items [0].revents & ZMQ_POLLIN) {
+ && items [0].revents & ZMQ_POLLIN
+ && items [1].revents & ZMQ_POLLOUT) {
  rc = forward(frontend_, backend_, capture_,msg);
  if (unlikely (rc < 0))
  return -1;
  }
  // Process a reply
  if (state == active
- && items [1].revents & ZMQ_POLLIN) {
+ && items [1].revents & ZMQ_POLLIN
+ && items [0].revents & ZMQ_POLLOUT) {
  rc = forward(backend_, frontend_, capture_,msg);
  if (unlikely (rc < 0))
  return -1;

tests/CMakeLists.txt

@@ -52,6 +52,7 @@ if(NOT WIN32)
  test_reqrep_ipc
  test_abstract_ipc
  test_proxy
+ test_proxy_terminate
  test_filter_ipc
  )
  if(HAVE_FORK)

tests/test_proxy_terminate.cpp

@@ -0,0 +1,113 @@
+/*
+ Copyright (c) 2007-2015 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 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
+ GNU Lesser General Public License for more details.
+
+ 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 "testutil.hpp"
+#include "../include/zmq_utils.h"
+
+// This is a test for issue #1382. The server thread creates a SUB-PUSH
+// steerable proxy. The main process then sends messages to the SUB
+// but there is no pull on the other side, previously the proxy blocks
+// in writing to the backend, preventing the proxy from terminating
+
+void
+server_task (void *ctx)
+{
+ // Frontend socket talks to main process
+ void *frontend = zmq_socket (ctx, ZMQ_SUB);
+ assert (frontend);
+ int rc = zmq_setsockopt (frontend, ZMQ_SUBSCRIBE, "", 0);
+ assert (rc == 0);
+ rc = zmq_bind (frontend, "tcp://127.0.0.1:15564");
+ assert (rc == 0);
+
+ // Nice socket which is never read
+ void *backend = zmq_socket (ctx, ZMQ_PUSH);
+ assert (backend);
+ rc = zmq_bind (backend, "tcp://127.0.0.1:15563");
+ assert (rc == 0);
+
+ // Control socket receives terminate command from main over inproc
+ void *control = zmq_socket (ctx, ZMQ_SUB);
+ assert (control);
+ rc = zmq_setsockopt (control, ZMQ_SUBSCRIBE, "", 0);
+ assert (rc == 0);
+ rc = zmq_connect (control, "inproc://control");
+ assert (rc == 0);
+
+ // Connect backend to frontend via a proxy
+ zmq_proxy_steerable (frontend, backend, NULL, control);
+
+ rc = zmq_close (frontend);
+ assert (rc == 0);
+ rc = zmq_close (backend);
+ assert (rc == 0);
+ rc = zmq_close (control);
+ assert (rc == 0);
+}
+
+
+// The main thread simply starts a basic steerable proxy server, publishes some messages, and then
+// waits for the server to terminate.
+
+int main (void)
+{
+ setup_test_environment ();
+
+ void *ctx = zmq_ctx_new ();
+ assert (ctx);
+ // Control socket receives terminate command from main over inproc
+ void *control = zmq_socket (ctx, ZMQ_PUB);
+ assert (control);
+ int rc = zmq_bind (control, "inproc://control");
+ assert (rc == 0);
+
+ void *thread = zmq_threadstart(&server_task, ctx);
+ msleep (500); // Run for 500 ms
+
+ // Start a secondary publisher which writes data to the SUB-PUSH server socket
+ void *publisher = zmq_socket (ctx, ZMQ_PUB);
+ assert (publisher);
+ rc = zmq_connect (publisher, "tcp://127.0.0.1:15564");
+ assert (rc == 0);
+
+ msleep (50);
+ rc = zmq_send (publisher, "This is a test", 14, 0);
+ assert (rc == 14);
+
+ msleep (50);
+ rc = zmq_send (publisher, "This is a test", 14, 0);
+ assert (rc == 14);
+
+ msleep (50);
+ rc = zmq_send (publisher, "This is a test", 14, 0);
+ assert (rc == 14);
+ rc = zmq_send (control, "TERMINATE", 9, 0);
+ assert (rc == 9);
+
+ rc = zmq_close (publisher);
+ assert (rc == 0);
+ rc = zmq_close (control);
+ assert (rc == 0);
+
+ zmq_threadclose (thread);
+
+ rc = zmq_ctx_term (ctx);
+ assert (rc == 0);
+ return 0;
+}