Merge branch 'master' of https://github.com/zeromq/libzmq

include/zmq.h

@@ -147,6 +147,7 @@ ZMQ_EXPORT int zmq_getmsgopt (zmq_msg_t *msg, int option, void *optval,
 /******************************************************************************/
 
 ZMQ_EXPORT void *zmq_init (int io_threads);
+ZMQ_EXPORT void *zmq_init_thread_safe (int io_threads);
 ZMQ_EXPORT int zmq_term (void *context);
 
 /******************************************************************************/

src/ctx.cpp

@@ -81,6 +81,16 @@ zmq::ctx_t::ctx_t (uint32_t io_threads_) :
     zmq_assert (rc == 0);
 }
 
+void zmq::ctx_t::set_thread_safe() 
+{
+  thread_safe_flag = true;
+}
+
+bool zmq::ctx_t::get_thread_safe() const
+{
+  return thread_safe_flag;
+}
+
 bool zmq::ctx_t::check_tag ()
 {
     return tag == 0xbadcafe0;

src/ctx.hpp

@@ -99,6 +99,10 @@ namespace zmq
             reaper_tid = 1
         };
 
+        // create thread safe sockets
+        void set_thread_safe();
+        bool get_thread_safe() const;
+
         ~ctx_t ();
     private:
 
@@ -151,6 +155,8 @@ namespace zmq
         zmq::socket_base_t *log_socket;
         mutex_t log_sync;
 
+        bool thread_safe_flag;
+
         ctx_t (const ctx_t&);
         const ctx_t &operator = (const ctx_t&);
     };

src/mtrie.cpp

@@ -34,6 +34,7 @@
 #include "mtrie.hpp"
 
 zmq::mtrie_t::mtrie_t () :
+    pipes (0),
     min (0),
     count (0),
     live_nodes (0)
@@ -42,6 +43,11 @@ zmq::mtrie_t::mtrie_t () :
 
 zmq::mtrie_t::~mtrie_t ()
 {
+    if (pipes) {
+        delete pipes;
+        pipes = 0;
+    }
+
     if (count == 1) {
         zmq_assert (next.node);
         delete next.node;
@@ -65,8 +71,10 @@ bool zmq::mtrie_t::add_helper (unsigned char *prefix_, size_t size_,
 {
     //  We are at the node corresponding to the prefix. We are done.
     if (!size_) {
-        bool result = pipes.empty ();
-        pipes.insert (pipe_);
+        bool result = !pipes;
+        if (!pipes)
+            pipes = new pipes_t;
+        pipes->insert (pipe_);
         return result;
     }
 
@@ -154,8 +162,11 @@ void zmq::mtrie_t::rm_helper (pipe_t *pipe_, unsigned char **buff_,
     void *arg_)
 {
     //  Remove the subscription from this node.
-    if (pipes.erase (pipe_) && pipes.empty ())
+    if (pipes && pipes->erase (pipe_) && pipes->empty ()) {
         func_ (*buff_, buffsize_, arg_);
+        delete pipes;
+        pipes = 0;
+    }
 
     //  Adjust the buffer.
     if (buffsize_ >= maxbuffsize_) {
@@ -207,9 +218,15 @@ bool zmq::mtrie_t::rm_helper (unsigned char *prefix_, size_t size_,
     pipe_t *pipe_)
 {
     if (!size_) {
-        pipes_t::size_type erased = pipes.erase (pipe_);
-        zmq_assert (erased == 1);
-        return pipes.empty ();
+        if (pipes) {
+            pipes_t::size_type erased = pipes->erase (pipe_);
+            zmq_assert (erased == 1);
+            if (pipes->empty ()) {
+                delete pipes;
+                pipes = 0;
+            }
+        }
+        return !pipes;
     }
 
     unsigned char c = *prefix_;
@@ -245,9 +262,11 @@ void zmq::mtrie_t::match (unsigned char *data_, size_t size_,
     while (true) {
 
         //  Signal the pipes attached to this node.
-        for (pipes_t::iterator it = current->pipes.begin ();
-              it != current->pipes.end (); ++it)
-            func_ (*it, arg_);
+        if (current->pipes) {
+            for (pipes_t::iterator it = current->pipes->begin ();
+                  it != current->pipes->end (); ++it)
+                func_ (*it, arg_);
+        }
 
         //  If we are at the end of the message, there's nothing more to match.
         if (!size_)
@@ -281,5 +300,5 @@ void zmq::mtrie_t::match (unsigned char *data_, size_t size_,
 
 bool zmq::mtrie_t::is_redundant () const
 {
-    return pipes.empty () && live_nodes == 0;
+    return !pipes && live_nodes == 0;
 }

src/mtrie.hpp

@@ -73,7 +73,7 @@ namespace zmq
         bool is_redundant () const;
 
         typedef std::set <zmq::pipe_t*> pipes_t;
-        pipes_t pipes;
+        pipes_t *pipes;
 
         unsigned char min;
         unsigned short count;

src/pair.cpp

@@ -36,7 +36,7 @@ zmq::pair_t::~pair_t ()
     zmq_assert (!pipe);
 }
 
-void zmq::pair_t::xattach_pipe (pipe_t *pipe_)
+void zmq::pair_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (!pipe);
     pipe = pipe_;

src/pair.hpp

@@ -42,7 +42,7 @@ namespace zmq
         ~pair_t ();
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xsend (zmq::msg_t *msg_, int flags_);
         int xrecv (zmq::msg_t *msg_, int flags_);
         bool xhas_in ();

src/pgm_receiver.cpp

@@ -117,8 +117,15 @@ void zmq::pgm_receiver_t::activate_in ()
     //  processed the whole buffer but failed to write
     //  the last message into the pipe.
     if (pending_bytes == 0) {
-        if (mru_decoder != NULL)
+        if (mru_decoder != NULL) {
             mru_decoder->process_buffer (NULL, 0);
+            session->flush ();
+        }
+
+        //  Resume polling.
+        set_pollin (pipe_handle);
+        set_pollin (socket_handle);
+
         return;
     }
 
@@ -128,6 +135,7 @@ void zmq::pgm_receiver_t::activate_in ()
     //  Ask the decoder to process remaining data.
     size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes);
     pending_bytes -= n;
+    session->flush ();
 
     if (pending_bytes > 0)
         return;
@@ -145,7 +153,8 @@ void zmq::pgm_receiver_t::in_event ()
     unsigned char *data = NULL;
     const pgm_tsi_t *tsi = NULL;
 
-    zmq_assert (pending_bytes == 0);
+    if (pending_bytes > 0)
+        return;
 
     if (has_rx_timer) {
         cancel_timer (rx_timer_id);

src/pgm_sender.cpp

@@ -91,16 +91,6 @@ void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
 
     //  Set POLLOUT for downlink_socket_handle.
     set_pollout (handle);
-
-    //  PGM is not able to pass subscriptions upstream, thus we have no idea
-    //  what messages are peers interested in. Because of that we have to
-    //  subscribe for all the messages.
-    msg_t msg;
-    msg.init_size (1);
-    *(unsigned char*) msg.data () = 1;
-    bool ok = session_->write (&msg);
-    zmq_assert (ok);
-    session_->flush ();
 }
 
 void zmq::pgm_sender_t::unplug ()

src/pull.cpp

@@ -34,7 +34,7 @@ zmq::pull_t::~pull_t ()
 {
 }
 
-void zmq::pull_t::xattach_pipe (pipe_t *pipe_)
+void zmq::pull_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
     fq.attach (pipe_);

src/pull.hpp

@@ -45,7 +45,7 @@ namespace zmq
     protected:
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xrecv (zmq::msg_t *msg_, int flags_);
         bool xhas_in ();
         void xread_activated (zmq::pipe_t *pipe_);

src/push.cpp

@@ -34,7 +34,7 @@ zmq::push_t::~push_t ()
 {
 }
 
-void zmq::push_t::xattach_pipe (pipe_t *pipe_)
+void zmq::push_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
     lb.attach (pipe_);

src/push.hpp

@@ -45,7 +45,7 @@ namespace zmq
     protected:
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xsend (zmq::msg_t *msg_, int flags_);
         bool xhas_out ();
         void xwrite_activated (zmq::pipe_t *pipe_);

src/socket_base.cpp

@@ -121,7 +121,8 @@ zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_) :
     destroyed (false),
     last_tsc (0),
     ticks (0),
-    rcvmore (false)
+    rcvmore (false),
+    thread_safe_flag (false)
 {
 }
 
@@ -209,14 +210,14 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_)
     return 0;
 }
 
-void zmq::socket_base_t::attach_pipe (pipe_t *pipe_)
+void zmq::socket_base_t::attach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     //  First, register the pipe so that we can terminate it later on.
     pipe_->set_event_sink (this);
     pipes.push_back (pipe_);
     
     //  Let the derived socket type know about new pipe.
-    xattach_pipe (pipe_);
+    xattach_pipe (pipe_, icanhasall_);
 
     //  If the socket is already being closed, ask any new pipes to terminate
     //  straight away.
@@ -458,8 +459,14 @@ int zmq::socket_base_t::connect (const char *addr_)
     rc = pipepair (parents, pipes, hwms, delays);
     errno_assert (rc == 0);
 
+    //  PGM does not support subscription forwarding; ask for all data to be
+    //  sent to this pipe.
+    bool icanhasall = false;
+    if (protocol == "pgm" || protocol == "epgm")
+        icanhasall = true;
+
     //  Attach local end of the pipe to the socket object.
-    attach_pipe (pipes [0]);
+    attach_pipe (pipes [0], icanhasall);
 
     //  Attach remote end of the pipe to the session object later on.
     session->attach_pipe (pipes [1]);
@@ -871,3 +878,18 @@ void zmq::socket_base_t::extract_flags (msg_t *msg_)
     rcvmore = msg_->flags () & msg_t::more ? true : false;
 }
 
+void zmq::socket_base_t::set_thread_safe()
+{
+   thread_safe_flag = true;
+}
+
+void zmq::socket_base_t::lock() 
+{
+   sync.lock();
+}
+
+void zmq::socket_base_t::unlock() 
+{
+   sync.unlock();
+}
+

src/socket_base.hpp

@@ -95,7 +95,10 @@ namespace zmq
         void write_activated (pipe_t *pipe_);
         void hiccuped (pipe_t *pipe_);
         void terminated (pipe_t *pipe_);
-
+        bool thread_safe() const { return thread_safe_flag; }
+        void set_thread_safe(); // should be in constructor, here for compat
+        void lock();
+        void unlock();
     protected:
 
         socket_base_t (zmq::ctx_t *parent_, uint32_t tid_);
@@ -103,7 +106,8 @@ namespace zmq
 
         //  Concrete algorithms for the x- methods are to be defined by
         //  individual socket types.
-        virtual void xattach_pipe (zmq::pipe_t *pipe_) = 0;
+        virtual void xattach_pipe (zmq::pipe_t *pipe_,
+            bool icanhasall_ = false) = 0;
 
         //  The default implementation assumes there are no specific socket
         //  options for the particular socket type. If not so, overload this
@@ -158,7 +162,7 @@ namespace zmq
         int check_protocol (const std::string &protocol_);
 
         //  Register the pipe with this socket.
-        void attach_pipe (zmq::pipe_t *pipe_);
+        void attach_pipe (zmq::pipe_t *pipe_, bool icanhasall_ = false);
 
         //  Processes commands sent to this socket (if any). If timeout is -1,
         //  returns only after at least one command was processed.
@@ -194,6 +198,8 @@ namespace zmq
 
         socket_base_t (const socket_base_t&);
         const socket_base_t &operator = (const socket_base_t&);
+        bool thread_safe_flag;
+        mutex_t sync;
     };
 
 }

src/xpub.cpp

@@ -37,11 +37,16 @@ zmq::xpub_t::~xpub_t ()
 {
 }
 
-void zmq::xpub_t::xattach_pipe (pipe_t *pipe_)
+void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
     dist.attach (pipe_);
 
+    //  If icanhasall_ is specified, the caller would like to subscribe
+    //  to all data on this pipe, implicitly.
+    if (icanhasall_)
+        subscriptions.add (NULL, 0, pipe_);
+
     //  The pipe is active when attached. Let's read the subscriptions from
     //  it, if any.
     xread_activated (pipe_);
@@ -51,14 +56,11 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
 {
     //  There are some subscriptions waiting. Let's process them.
     msg_t sub;
-    sub.init ();
     while (true) {
 
         //  Grab next subscription.
-        if (!pipe_->read (&sub)) {
-            sub.close ();
+        if (!pipe_->read (&sub))
             return;
-        }
 
         //  Apply the subscription to the trie.
         unsigned char *data = (unsigned char*) sub.data ();
@@ -76,6 +78,8 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
                 pending.push_back (blob_t ((unsigned char*) sub.data (),
                     sub.size ()));
         }
+
+        sub.close();
     }
 }
 

src/xpub.hpp

@@ -47,7 +47,7 @@ namespace zmq
         ~xpub_t ();
 
         //  Implementations of virtual functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_ = false);
         int xsend (zmq::msg_t *msg_, int flags_);
         bool xhas_out ();
         int xrecv (zmq::msg_t *msg_, int flags_);

src/xrep.cpp

@@ -55,7 +55,7 @@ zmq::xrep_t::~xrep_t ()
     prefetched_msg.close ();
 }
 
-void zmq::xrep_t::xattach_pipe (pipe_t *pipe_)
+void zmq::xrep_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
 

src/xrep.hpp

@@ -48,7 +48,7 @@ namespace zmq
         ~xrep_t ();
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xsend (msg_t *msg_, int flags_);
         int xrecv (msg_t *msg_, int flags_);
         bool xhas_in ();

src/xreq.cpp

@@ -46,7 +46,7 @@ zmq::xreq_t::~xreq_t ()
     prefetched_msg.close ();
 }
 
-void zmq::xreq_t::xattach_pipe (pipe_t *pipe_)
+void zmq::xreq_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
     fq.attach (pipe_);

src/xreq.hpp

@@ -46,7 +46,7 @@ namespace zmq
     protected:
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xsend (zmq::msg_t *msg_, int flags_);
         int xrecv (zmq::msg_t *msg_, int flags_);
         bool xhas_in ();

src/xsub.cpp

@@ -45,7 +45,7 @@ zmq::xsub_t::~xsub_t ()
     errno_assert (rc == 0);
 }
 
-void zmq::xsub_t::xattach_pipe (pipe_t *pipe_)
+void zmq::xsub_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_)
 {
     zmq_assert (pipe_);
     fq.attach (pipe_);

src/xsub.hpp

@@ -45,7 +45,7 @@ namespace zmq
     protected:
 
         //  Overloads of functions from socket_base_t.
-        void xattach_pipe (zmq::pipe_t *pipe_);
+        void xattach_pipe (zmq::pipe_t *pipe_, bool icanhasall_);
         int xsend (zmq::msg_t *msg_, int flags_);
         bool xhas_out ();
         int xrecv (zmq::msg_t *msg_, int flags_);

src/zmq.cpp

@@ -78,6 +78,8 @@
 typedef char check_msg_t_size
     [sizeof (zmq::msg_t) ==  sizeof (zmq_msg_t) ? 1 : -1];
 
+// Version.
+
 void zmq_version (int *major_, int *minor_, int *patch_)
 {
     *major_ = ZMQ_VERSION_MAJOR;
@@ -85,12 +87,21 @@ void zmq_version (int *major_, int *minor_, int *patch_)
     *patch_ = ZMQ_VERSION_PATCH;
 }
 
+// Errors.
+
 const char *zmq_strerror (int errnum_)
 {
     return zmq::errno_to_string (errnum_);
 }
 
-void *zmq_init (int io_threads_)
+int zmq_errno ()
+{
+    return errno;
+}
+
+// Contexts.
+
+static zmq::ctx_t *inner_init (int io_threads_)
 {
     if (io_threads_ < 0) {
         errno = EINVAL;
@@ -139,7 +150,19 @@ void *zmq_init (int io_threads_)
     //  Create 0MQ context.
     zmq::ctx_t *ctx = new (std::nothrow) zmq::ctx_t ((uint32_t) io_threads_);
     alloc_assert (ctx);
-    return (void*) ctx;
+    return ctx;
+}
+
+void *zmq_init (int io_threads_)
+{
+    return (void*) inner_init (io_threads_);
+}
+
+void *zmq_init_thread_safe (int io_threads_)
+{
+  zmq::ctx_t *ctx = inner_init (io_threads_);
+  ctx->set_thread_safe();
+  return (void*) ctx;
 }
 
 int zmq_term (void *ctx_)
@@ -168,13 +191,18 @@ int zmq_term (void *ctx_)
     return rc;
 }
 
+// Sockets.
+
 void *zmq_socket (void *ctx_, int type_)
 {
     if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
         errno = EFAULT;
         return NULL;
     }
-    return (void*) (((zmq::ctx_t*) ctx_)->create_socket (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;
 }
 
 int zmq_close (void *s_)
@@ -194,8 +222,11 @@ int zmq_setsockopt (void *s_, int option_, const void *optval_,
         errno = ENOTSOCK;
         return -1;
     }
-    return (((zmq::socket_base_t*) s_)->setsockopt (option_, optval_,
-        optvallen_));
+    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;
 }
 
 int zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_)
@@ -204,8 +235,11 @@ int zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_)
         errno = ENOTSOCK;
         return -1;
     }
-    return (((zmq::socket_base_t*) s_)->getsockopt (option_, optval_,
-        optvallen_));
+    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;
 }
 
 int zmq_bind (void *s_, const char *addr_)
@@ -214,7 +248,11 @@ int zmq_bind (void *s_, const char *addr_)
         errno = ENOTSOCK;
         return -1;
     }
-    return (((zmq::socket_base_t*) s_)->bind (addr_));
+    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;
 }
 
 int zmq_connect (void *s_, const char *addr_)
@@ -223,7 +261,35 @@ int zmq_connect (void *s_, const char *addr_)
         errno = ENOTSOCK;
         return -1;
     }
-    return (((zmq::socket_base_t*) s_)->connect (addr_));
+    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;
+}
+
+// Sending functions.
+
+static int inner_sendmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
+{
+    int sz = (int) zmq_msg_size (msg_);
+    int rc = s_->send ((zmq::msg_t*) msg_, flags_);
+    if (unlikely (rc < 0))
+        return -1;
+    return sz;
+}
+
+int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+    int result = inner_sendmsg (s, msg_, flags_);
+    if(s->thread_safe()) s->unlock();
+    return result;
 }
 
 int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
@@ -234,7 +300,10 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
         return -1;
     memcpy (zmq_msg_data (&msg), buf_, len_);
 
-    rc = zmq_sendmsg (s_, &msg, flags_);
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+    rc = inner_sendmsg (s, &msg, flags_);
+    if(s->thread_safe()) s->unlock();
     if (unlikely (rc < 0)) {
         int err = errno;
         int rc2 = zmq_msg_close (&msg);
@@ -248,13 +317,44 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
     return rc;
 }
 
+// Receiving functions.
+
+static int inner_recvmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
+{
+    int rc = s_->recv ((zmq::msg_t*) msg_, flags_);
+    if (unlikely (rc < 0))
+        return -1;
+    return (int) zmq_msg_size (msg_);
+}
+
+int zmq_recvmsg (void *s_, zmq_msg_t *msg_, int flags_)
+{
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+    int result = inner_recvmsg(s, msg_, flags_);
+    if(s->thread_safe()) s->unlock();
+    return result;
+}
+
+
 int zmq_recv (void *s_, void *buf_, size_t len_, int flags_)
 {
+    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
+        errno = ENOTSOCK;
+        return -1;
+    }
     zmq_msg_t msg;
     int rc = zmq_msg_init (&msg);
     errno_assert (rc == 0);
 
-    int nbytes = zmq_recvmsg (s_, &msg, flags_);
+    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;
+    if(s->thread_safe()) s->lock();
+    int nbytes = inner_recvmsg (s, &msg, flags_);
+    if(s->thread_safe()) s->unlock();
     if (unlikely (nbytes < 0)) {
         int err = errno;
         rc = zmq_msg_close (&msg);
@@ -274,30 +374,7 @@ int zmq_recv (void *s_, void *buf_, size_t len_, int flags_)
     return nbytes;    
 }
 
-int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_)
-{
-    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
-        errno = ENOTSOCK;
-        return -1;
-    }
-    int sz = (int) zmq_msg_size (msg_);
-    int rc = (((zmq::socket_base_t*) s_)->send ((zmq::msg_t*) msg_, flags_));
-    if (unlikely (rc < 0))
-        return -1;
-    return sz;
-}
-
-int zmq_recvmsg (void *s_, zmq_msg_t *msg_, int flags_)
-{
-    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {
-        errno = ENOTSOCK;
-        return -1;
-    }
-    int rc = (((zmq::socket_base_t*) s_)->recv ((zmq::msg_t*) msg_, flags_));
-    if (unlikely (rc < 0))
-        return -1;
-    return (int) zmq_msg_size (msg_);
-}
+// Message manipulators.
 
 int zmq_msg_init (zmq_msg_t *msg_)
 {
@@ -359,6 +436,8 @@ int zmq_getmsgopt (zmq_msg_t *msg_, int option_, void *optval_,
     }
 }
 
+// Polling.
+
 int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
 {
 #if defined ZMQ_POLL_BASED_ON_POLL
@@ -711,11 +790,6 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
 #endif
 }
 
-int zmq_errno ()
-{
-    return errno;
-}
-
 #if defined ZMQ_POLL_BASED_ON_SELECT
 #undef ZMQ_POLL_BASED_ON_SELECT
 #endif

tests/Makefile.am

@@ -17,6 +17,7 @@ if !ON_MINGW
 noinst_PROGRAMS += test_shutdown_stress \
                    test_pair_ipc \
                    test_reqrep_ipc \
+                   test_ts_context \
                    test_timeo
 endif
 
@@ -35,6 +36,7 @@ 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

@@ -0,0 +1,129 @@
+/*
+    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 ;
+}