Merge pull request #1525 from somdoron/master

problem: user cannot poll on thread safe sockets

.gitignore

@@ -112,6 +112,9 @@ test_client_drop_more
 test_client_server
 test_server_drop_more
 test_thread_safe
+test_thread_safe_polling
+test_getsockopt_memset
+test_stream_exceeds_buffer
 tests/test*.log
 tests/test*.trs
 src/platform.hpp*

Makefile.am

@@ -366,7 +366,8 @@ test_apps = \
 	tests/test_thread_safe \
 	tests/test_socketopt_hwm \
 	tests/test_heartbeats \
-	tests/test_stream_exceeds_buffer
+	tests/test_stream_exceeds_buffer \
+	tests/test_thread_safe_polling
 
 tests_test_system_SOURCES = tests/test_system.cpp
 tests_test_system_LDADD = src/libzmq.la
@@ -577,6 +578,10 @@ tests_test_heartbeats_LDADD = src/libzmq.la
 tests_test_stream_exceeds_buffer_SOURCES = tests/test_stream_exceeds_buffer.cpp
 tests_test_stream_exceeds_buffer_LDADD = src/libzmq.la
 
+tests_test_thread_safe_polling_SOURCES = tests/test_thread_safe_polling.cpp
+tests_test_thread_safe_polling_LDADD = src/libzmq.la
+
+
 if !ON_MINGW
 if !ON_CYGWIN
 test_apps += \

include/zmq.h

@@ -322,6 +322,7 @@ ZMQ_EXPORT uint32_t zmq_msg_get_routing_id(zmq_msg_t *msg);
 #define ZMQ_XPUB_VERBOSE_UNSUBSCRIBE 78
 #define ZMQ_CONNECT_TIMEOUT 79
 #define ZMQ_TCP_RETRANSMIT_TIMEOUT 80
+#define ZMQ_THREAD_SAFE 81
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1
@@ -382,7 +383,8 @@ ZMQ_EXPORT int zmq_send (void *s, const void *buf, size_t len, int flags);
 ZMQ_EXPORT int zmq_send_const (void *s, const void *buf, size_t len, int flags);
 ZMQ_EXPORT int zmq_recv (void *s, void *buf, size_t len, int flags);
 ZMQ_EXPORT int zmq_socket_monitor (void *s, const char *addr, int events);
-
+ZMQ_EXPORT int zmq_add_poller (void *s, void *p);
+ZMQ_EXPORT int zmq_remove_poller (void *s, void *p);
 
 /******************************************************************************/
 /*  I/O multiplexing.                                                         */
@@ -396,6 +398,7 @@ ZMQ_EXPORT int zmq_socket_monitor (void *s, const char *addr, int events);
 typedef struct zmq_pollitem_t
 {
     void *socket;
+    void *poller;
 #if defined _WIN32
     SOCKET fd;
 #else
@@ -407,7 +410,9 @@ typedef struct zmq_pollitem_t
 
 #define ZMQ_POLLITEMS_DFLT 16
 
-ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
+ZMQ_EXPORT int  zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
+ZMQ_EXPORT void *zmq_poller_new ();
+ZMQ_EXPORT int  zmq_poller_close (void *p);
 
 /******************************************************************************/
 /*  Message proxying                                                          */

src/socket_base.cpp

@@ -423,11 +423,56 @@ int zmq::socket_base_t::getsockopt (int option_, void *optval_,
         return 0;
     }
 
+    if (option_ == ZMQ_THREAD_SAFE) {
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            EXIT_MUTEX();
+            return -1;
+        }
+        memset(optval_, 0, *optvallen_);
+        *((int*) optval_) = thread_safe	? 1 : 0;
+        *optvallen_ = sizeof (int);
+        EXIT_MUTEX();
+        return 0;
+    }  
+
     int rc = options.getsockopt (option_, optval_, optvallen_);
     EXIT_MUTEX();
     return rc;
 }
 
+int zmq::socket_base_t::add_signaler(signaler_t *s_)
+{
+    ENTER_MUTEX();
+
+    if (!thread_safe) {
+        errno = EINVAL;
+        EXIT_MUTEX();
+        return -1;  
+    }
+
+    ((mailbox_safe_t*)mailbox)->add_signaler(s_);
+
+    EXIT_MUTEX();
+    return 0;
+}
+
+int zmq::socket_base_t::remove_signaler(signaler_t *s_)
+{
+    ENTER_MUTEX();
+
+    if (!thread_safe) {
+        errno = EINVAL;
+        EXIT_MUTEX();
+        return -1;
+    }
+
+    ((mailbox_safe_t*)mailbox)->remove_signaler(s_);
+
+    EXIT_MUTEX();
+    return 0;
+}
+
 int zmq::socket_base_t::bind (const char *addr_)
 {
     ENTER_MUTEX();

src/socket_base.hpp

@@ -90,6 +90,8 @@ namespace zmq
         int term_endpoint (const char *addr_);
         int send (zmq::msg_t *msg_, int flags_);
         int recv (zmq::msg_t *msg_, int flags_);
+        int add_signaler (signaler_t *s);
+        int remove_signaler (signaler_t *s);
         int close ();
 
         //  These functions are used by the polling mechanism to determine

src/zmq.cpp

@@ -74,6 +74,7 @@ struct iovec {
 #include "msg.hpp"
 #include "fd.hpp"
 #include "metadata.hpp"
+#include "signaler.hpp"
 
 #if !defined ZMQ_HAVE_WINDOWS
 #include <unistd.h>
@@ -561,6 +562,34 @@ int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_)
     return nread;
 }
 
+// Add/remove poller from a socket
+
+int zmq_add_poller (void *s_, void *p_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    zmq::signaler_t *p = (zmq::signaler_t *) p_;
+
+    return s->add_signaler(p);
+}
+
+int zmq_remove_poller (void *s_, void *p_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    zmq::signaler_t *p = (zmq::signaler_t *) p_;
+
+    return s->remove_signaler(p);
+}
+
+
+
 // Message manipulators.
 
 int zmq_msg_init (zmq_msg_t *msg_)
@@ -680,6 +709,31 @@ const char *zmq_msg_gets (zmq_msg_t *msg_, const char *property_)
     }
 }
 
+// Create poller 
+
+void *zmq_poller_new () 
+{
+    return new zmq::signaler_t ();
+}
+
+// Close poller
+
+int zmq_poller_close (void* p)
+{
+    zmq::signaler_t *s = (zmq::signaler_t*)p;
+    delete s;
+
+    return 0;	
+}
+
+// Get poller fd
+
+zmq::fd_t zmq_poller_get_fd (void *p)
+{
+    zmq::signaler_t *s = (zmq::signaler_t*)p;
+    return s->get_fd ();	
+}
+
 // Polling.
 
 int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
@@ -725,14 +779,36 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
         //  If the poll item is a 0MQ socket, we poll on the file descriptor
         //  retrieved by the ZMQ_FD socket option.
         if (items_ [i].socket) {
-            size_t zmq_fd_size = sizeof (zmq::fd_t);
-            if (zmq_getsockopt (items_ [i].socket, ZMQ_FD, &pollfds [i].fd,
-                &zmq_fd_size) == -1) {
+            int thread_safe;
+            size_t thread_safe_size = sizeof(int);
+
+            if (zmq_getsockopt (items_ [i].socket, ZMQ_THREAD_SAFE, &thread_safe, 
+                &thread_safe_size) == -1) {
                 if (pollfds != spollfds)
                     free (pollfds);
                 return -1;
             }
-            pollfds [i].events = items_ [i].events ? POLLIN : 0;
+
+            if (thread_safe) {
+                if (!items_ [i].poller) {
+                    if (pollfds != spollfds)
+                        free (pollfds);
+                    errno = EINVAL;
+                    return -1;
+                }
+
+                pollfds [i].fd = zmq_poller_get_fd (items_ [i].poller);
+            }
+            else {
+                size_t zmq_fd_size = sizeof (zmq::fd_t);
+                if (zmq_getsockopt (items_ [i].socket, ZMQ_FD, &pollfds [i].fd,
+                    &zmq_fd_size) == -1) {
+                    if (pollfds != spollfds)
+                        free (pollfds);
+                    return -1;
+                }
+            }
+            pollfds [i].events = items_ [i].events ? POLLIN : 0;            
         }
         //  Else, the poll item is a raw file descriptor. Just convert the
         //  events to normal POLLIN/POLLOUT for poll ().
@@ -888,11 +964,29 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
         //  If the poll item is a 0MQ socket we are interested in input on the
         //  notification file descriptor retrieved by the ZMQ_FD socket option.
         if (items_ [i].socket) {
-            size_t zmq_fd_size = sizeof (zmq::fd_t);
-            zmq::fd_t notify_fd;
-            if (zmq_getsockopt (items_ [i].socket, ZMQ_FD, &notify_fd,
-                &zmq_fd_size) == -1)
+            int thread_safe;
+            size_t thread_safe_size = sizeof(int);
+
+            if (zmq_getsockopt (items_ [i].socket, ZMQ_THREAD_SAFE, &thread_safe, 
+                &thread_safe_size) == -1) 
                 return -1;
+
+            zmq::fd_t notify_fd;
+
+            if (thread_safe) {
+                if (!items_ [i].poller) {
+                    errno = EINVAL;
+                    return -1;
+                }
+
+                notify_fd = zmq_poller_get_fd (items_ [i].poller);
+            }
+            else {
+                size_t zmq_fd_size = sizeof (zmq::fd_t);
+                if (zmq_getsockopt (items_ [i].socket, ZMQ_FD, &notify_fd,
+                    &zmq_fd_size) == -1)
+                    return -1;
+            }
             if (items_ [i].events) {
                 FD_SET (notify_fd, &pollset_in);
                 if (maxfd < notify_fd)

tests/CMakeLists.txt

@@ -52,6 +52,7 @@ set(tests
         test_client_server
 	test_sockopt_hwm
         test_heartbeats
+        test_thread_safe_polling
 )
 if(NOT WIN32)
   list(APPEND tests

tests/test_thread_safe.cpp

@@ -41,6 +41,13 @@ int main (void)
     void *client = zmq_socket (ctx, ZMQ_CLIENT);
     void *client2 = zmq_socket (ctx, ZMQ_CLIENT);
 
+    int thread_safe;
+    size_t size = sizeof(int);
+
+    zmq_getsockopt (client, ZMQ_THREAD_SAFE, &thread_safe, &size);
+
+    assert (thread_safe == 1);
+
     int rc;
 
     rc = zmq_bind (client, "tcp://127.0.0.1:5560");

tests/test_thread_safe_polling.cpp

+/*
+    Copyright (c) 2007-2015 Contributors as noted in the AUTHORS file
+
+    This file is part of libzmq, the ZeroMQ core engine in C++.
+
+    libzmq is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License (LGPL) as published
+    by the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    As a special exception, the Contributors give you permission to link
+    this library with independent modules to produce an executable,
+    regardless of the license terms of these independent modules, and to
+    copy and distribute the resulting executable under terms of your choice,
+    provided that you also meet, for each linked independent module, the
+    terms and conditions of the license of that module. An independent
+    module is a module which is not derived from or based on this library.
+    If you modify this library, you must extend this exception to your
+    version of the library.
+
+    libzmq 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"
+
+void worker(void* s);
+
+int main (void)
+{
+    setup_test_environment();
+    void *ctx = zmq_ctx_new ();
+    assert (ctx);
+
+    void *server = zmq_socket (ctx, ZMQ_SERVER);
+    void *server2 = zmq_socket (ctx, ZMQ_SERVER);
+    void *poller = zmq_poller_new ();
+
+    int rc;
+
+    rc = zmq_add_poller (server, poller);
+    assert (rc == 0);
+
+    rc = zmq_add_poller (server2, poller);
+    assert (rc == 0);
+
+    zmq_pollitem_t items[2];
+
+    items[0].socket = server;
+    items[0].poller = poller;
+    items[0].events = ZMQ_POLLIN;
+
+    items[1].socket = server2;
+    items[1].poller = poller;
+    items[1].events = ZMQ_POLLIN;
+
+    rc = zmq_bind (server, "tcp://127.0.0.1:5560");
+    assert (rc == 0);
+    
+    rc = zmq_bind (server2, "tcp://127.0.0.1:5561");
+    assert (rc == 0);
+
+    void* t = zmq_threadstart(worker, ctx);
+
+    assert (rc == 0);    
+
+    rc = zmq_poll (items, 2, -1);
+    assert (rc == 1);
+
+    assert (items[0].revents == ZMQ_POLLIN);
+    assert (items[1].revents == 0);
+
+    zmq_msg_t msg;
+    rc = zmq_msg_init(&msg);    
+    rc = zmq_msg_recv(&msg, server, ZMQ_DONTWAIT);
+    assert (rc == 1);    
+
+    rc = zmq_poll (items, 2, -1);
+    assert (rc == 1);
+
+    assert (items[0].revents == 0); 
+    assert (items[1].revents == ZMQ_POLLIN);
+
+    rc = zmq_msg_recv(&msg, server2, ZMQ_DONTWAIT);
+    assert (rc == 1);
+
+    rc = zmq_poll (items, 2, 0);
+    assert (rc == 0);
+
+    assert (items[0].revents == 0); 
+    assert (items[1].revents == 0);
+
+    zmq_threadclose(t);
+
+    rc = zmq_msg_close(&msg);
+    assert (rc == 0);
+
+    rc = zmq_remove_poller (server, poller);
+    assert (rc == 0);
+
+    rc = zmq_remove_poller (server2, poller);
+    assert (rc == 0);
+
+    rc = zmq_poller_close (poller);
+    assert (rc == 0);
+
+    rc = zmq_close (server);
+    assert (rc == 0);
+
+    rc = zmq_close (server2);
+    assert (rc == 0);
+
+    rc = zmq_ctx_term (ctx);
+    assert (rc == 0);
+
+    return 0;
+}
+
+void worker(void* ctx)
+{
+    void *client = zmq_socket (ctx, ZMQ_CLIENT);
+
+    int rc = zmq_connect (client, "tcp://127.0.0.1:5560");
+    assert (rc == 0);     
+
+    msleep(100);
+
+    zmq_msg_t msg;
+    rc = zmq_msg_init_size(&msg,1);
+    assert (rc == 0);
+
+    char * data = (char *)zmq_msg_data(&msg);
+    data[0] = 1;
+
+    rc = zmq_msg_send(&msg, client, 0);
+    assert (rc == 1);
+
+    rc = zmq_disconnect (client, "tcp://127.0.0.1:5560");
+    assert (rc == 0);     
+
+    rc = zmq_connect (client, "tcp://127.0.0.1:5561");
+    assert (rc == 0);
+
+    msleep(100);
+
+    rc = zmq_msg_close(&msg);
+    assert (rc == 0);    
+
+    rc = zmq_msg_init_size(&msg,1);
+    assert (rc == 0);
+
+    data = (char *)zmq_msg_data(&msg);
+    data[0] = 1;
+
+    rc = zmq_msg_send(&msg, client, 0);
+    assert (rc == 1);
+
+    rc = zmq_msg_close(&msg);
+    assert (rc == 0);
+
+    rc = zmq_close (client);
+    assert (rc == 0);
+}
+
+