Merge pull request #258 from skaller/master

Remove thread safe socket code from C API and tests.

include/zmq.h

@@ -180,7 +180,6 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, const void *optval,
 /******************************************************************************/
 
 ZMQ_EXPORT zmq_ctx_t zmq_init (int io_threads);
-ZMQ_EXPORT zmq_ctx_t zmq_init_thread_safe (int io_threads);
 ZMQ_EXPORT int zmq_term (zmq_ctx_t context);
 
 /******************************************************************************/

src/zmq.cpp

@@ -172,13 +172,6 @@ void *zmq_init (int io_threads_)
     return (void *) s_init (io_threads_);
 }
 
-void *zmq_init_thread_safe (int io_threads_)
-{
-    zmq::ctx_t *ctx = s_init (io_threads_);
-    ctx->set_thread_safe();
-    return (void *) ctx;
-}
-
 int zmq_term (void *ctx_)
 {
     if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
@@ -214,7 +207,6 @@ void *zmq_socket (void *ctx_, int type_)
     }
     zmq::ctx_t *ctx = (zmq::ctx_t*) ctx_;
     zmq::socket_base_t *s = ctx->create_socket (type_);
-    if (ctx->get_thread_safe ()) s->set_thread_safe ();
     return (void *) s;
 }
 
@@ -236,11 +228,7 @@ int zmq_setsockopt (void *s_, int option_, const void *optval_,
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s->setsockopt (option_, optval_, optvallen_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 
@@ -251,11 +239,7 @@ int zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s->getsockopt (option_, optval_, optvallen_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 
@@ -266,11 +250,7 @@ int zmq_bind (void *s_, const char *addr_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s->bind (addr_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 
@@ -281,11 +261,7 @@ int zmq_connect (void *s_, const char *addr_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s->connect (addr_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 
@@ -320,11 +296,7 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
     memcpy (zmq_msg_data (&msg), buf_, len_);
 
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     rc = s_sendmsg (s, &msg, flags_);
-    if (s->thread_safe())
-        s->unlock();
     if (unlikely (rc < 0)) {
         int err = errno;
         int rc2 = zmq_msg_close (&msg);
@@ -353,8 +325,6 @@ int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_)
     int rc = 0;
     zmq_msg_t msg;
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     
     for (size_t i = 0; i < count_; ++i) {
         rc = zmq_msg_init_size (&msg, a_[i].iov_len);
@@ -375,8 +345,6 @@ int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_)
            break;
         }
     }
-    if (s->thread_safe())
-        s->unlock();
     return rc; 
 }
 
@@ -409,11 +377,7 @@ int zmq_recv (void *s_, void *buf_, size_t len_, int flags_)
     errno_assert (rc == 0);
 
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int nbytes = s_recvmsg (s, &msg, flags_);
-    if (s->thread_safe())
-        s->unlock();
     if (unlikely (nbytes < 0)) {
         int err = errno;
         rc = zmq_msg_close (&msg);
@@ -460,8 +424,6 @@ int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
 
     size_t count = (int) *count_;
     int nread = 0;
@@ -493,8 +455,6 @@ int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_)
         // Assume zmq_socket ZMQ_RVCMORE is properly set.
         recvmore = ((zmq::msg_t*) (void *) &msg)->flags () & zmq::msg_t::more;
     }
-    if (s->thread_safe())
-        s->unlock();
     return nread;
 }
 
@@ -523,11 +483,7 @@ int zmq_msg_send (zmq_msg_t *msg_, void *s_, int flags_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s_sendmsg (s, msg_, flags_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 
@@ -538,11 +494,7 @@ int zmq_msg_recv (zmq_msg_t *msg_, void *s_, int flags_)
         return -1;
     }
     zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
-    if (s->thread_safe())
-        s->lock();
     int result = s_recvmsg (s, msg_, flags_);
-    if (s->thread_safe())
-        s->unlock();
     return result;
 }
 

tests/Makefile.am

@@ -18,7 +18,6 @@ if !ON_MINGW
 noinst_PROGRAMS += test_shutdown_stress \
                    test_pair_ipc \
                    test_reqrep_ipc \
-                   test_ts_context \
                    test_timeo
 endif
 
@@ -38,7 +37,6 @@ test_shutdown_stress_SOURCES = test_shutdown_stress.cpp
 test_pair_ipc_SOURCES = test_pair_ipc.cpp testutil.hpp
 test_reqrep_ipc_SOURCES = test_reqrep_ipc.cpp testutil.hpp
 test_timeo_SOURCES = test_timeo.cpp
-test_ts_context_SOURCES = test_ts_context.cpp
 endif
 
 TESTS = $(noinst_PROGRAMS)

tests/test_ts_context.cpp

-/*
-    Copyright (c) 2010-2011 250bpm s.r.o.
-    Copyright (c) 2011 iMatix Corporation
-    Copyright (c) 2010-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 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 <assert.h>
-#include <stdio.h>
-#include <pthread.h>
-#include <string.h>
-
-#include "../include/zmq.h"
-#include "../include/zmq_utils.h"
-
-#define THREAD_COUNT 30
-#define NMESSAGES 20
-
-struct thread_data_t 
-{
-  int thread_index;
-  void *socket;
-  pthread_t pthr;
-};
-
-extern "C" 
-{
-  static void *source(void *client_data)
-  {
-      // Wait a bit util all threads created and  subscriber ready.
-      zmq_sleep(2); // ms
-
-      // Our thread number and socket.
-      thread_data_t *td = (thread_data_t *) client_data;
-
-      // Buffer for messages.
-      char buffer[20];
-      memset (buffer, 0, 20);
-
-      // Send messages.
-      for (int i = 0; i < NMESSAGES; ++i)
-      {
-        sprintf(buffer,"Th %02d count %02d", td->thread_index, i);   
-        int rc = zmq_send(td->socket,buffer,20,0);
-        assert (rc == 20);
-        zmq_sleep(1); // Don't overload the socket.
-      }
- 
-      return 0;
-  }
-}
-
-
-int main (int argc, char *argv [])
-{
-    fprintf (stderr, "test_ts_context running...\n");
-
-    // Make a thread safe context.
-    void *ctx = zmq_init_thread_safe (1);
-    assert (ctx);
-
-    //  Create a publisher.
-    void *pub = zmq_socket (ctx, ZMQ_PUB);
-    assert (pub);
-    int rc = zmq_bind (pub, "tcp://127.0.0.1:5560");
-    assert (rc == 0);
-
-    //  Create a subscriber.
-    void *sub = zmq_socket (ctx, ZMQ_SUB);
-    assert (sub);
-    rc = zmq_connect (sub, "tcp://127.0.0.1:5560");
-    assert (rc == 0);
-
-    //  Subscribe for all messages.
-    rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);
-    assert (rc == 0);
-
-    thread_data_t threads [THREAD_COUNT];
-    
-    // Create workers.
-    for(int i = 0; i < THREAD_COUNT; ++i)
-    {
-        threads [i].thread_index = i;
-        threads [i].socket = pub;
-        rc = pthread_create(&threads[i].pthr, NULL, source, threads+i);
-        assert (rc == 0);
-    }
-
-    // Gather all the Messages.
-    char buff [20];
-    for(int i= 1; i<=THREAD_COUNT * NMESSAGES; ++i)
-    {
-      rc = zmq_recv (sub, buff, 20, 0);
-      //fprintf (stderr, "%d/%d: %s\n",i,THREAD_COUNT * NMESSAGES, buff); // debug it
-      assert (rc >= 0);
-    }
-
-    // Wait for worker death.
-    for(int i = 0; i < THREAD_COUNT; ++i)
-    {
-        rc = pthread_join(threads[i].pthr, NULL);
-        assert (rc == 0);
-    }
-
-
-    //  Clean up.
-    rc = zmq_close (pub);
-    assert (rc == 0);
-    rc = zmq_close (sub);
-    assert (rc == 0);
-    rc = zmq_term (ctx);
-    assert (rc == 0);
-
-    return 0 ;
-}