issue 1 - Change zmq_term semantics

6 files changed, 91 insertions, 12 deletions

diff --git a/include/zmq.h b/include/zmq.h
index e067115..fc52584 100644
--- a/include/zmq.h
+++ b/include/zmq.h
@@ -83,6 +83,7 @@ ZMQ_EXPORT void zmq_version (int *major, int *minor, int *patch);
 #define EMTHREAD (ZMQ_HAUSNUMERO + 50)
 #define EFSM (ZMQ_HAUSNUMERO + 51)
 #define ENOCOMPATPROTO (ZMQ_HAUSNUMERO + 52)
+#define ETERM (ZMQ_HAUSNUMERO + 53)
 
 /*  This function retrieves the errno as it is known to 0MQ library. The goal */
 /*  of this function is to make the code 100% portable, including where 0MQ   */
diff --git a/src/app_thread.cpp b/src/app_thread.cpp
index 61e49e5..6141c06 100644
--- a/src/app_thread.cpp
+++ b/src/app_thread.cpp
@@ -62,7 +62,8 @@
 zmq::app_thread_t::app_thread_t (dispatcher_t *dispatcher_, int thread_slot_,
       int flags_) :
     object_t (dispatcher_, thread_slot_),
-    last_processing_time (0)
+    last_processing_time (0),
+    terminated (false)
 {
     if (flags_ & ZMQ_POLL) {
         signaler = new (std::nothrow) fd_signaler_t;
@@ -81,12 +82,17 @@ zmq::app_thread_t::~app_thread_t ()
     delete signaler;
 }
 
+void zmq::app_thread_t::stop ()
+{
+    send_stop ();
+}
+
 zmq::i_signaler *zmq::app_thread_t::get_signaler ()
 {
     return signaler;
 }
 
-void zmq::app_thread_t::process_commands (bool block_, bool throttle_)
+bool zmq::app_thread_t::process_commands (bool block_, bool throttle_)
 {
     uint64_t signals;
     if (block_)
@@ -117,7 +123,7 @@ void zmq::app_thread_t::process_commands (bool block_, bool throttle_)
             //  Check whether certain time have elapsed since last command
             //  processing.
             if (current_time - last_processing_time <= max_command_delay)
-                return;
+                return !terminated;
             last_processing_time = current_time;
         }
 #endif
@@ -138,6 +144,8 @@ void zmq::app_thread_t::process_commands (bool block_, bool throttle_)
             }
         }
     }
+
+    return !terminated;
 }
 
 zmq::socket_base_t *zmq::app_thread_t::create_socket (int type_)
@@ -190,3 +198,14 @@ void zmq::app_thread_t::remove_socket (socket_base_t *socket_)
     if (sockets.empty ())
         dispatcher->no_sockets (this);
 }
+
+void zmq::app_thread_t::process_stop ()
+{
+    terminated = true;
+}
+
+bool zmq::app_thread_t::is_terminated ()
+{
+    return terminated;
+}
+
diff --git a/src/app_thread.hpp b/src/app_thread.hpp
index 9bd5641..1c2f47a 100644
--- a/src/app_thread.hpp
+++ b/src/app_thread.hpp
@@ -38,14 +38,19 @@ namespace zmq
 
         ~app_thread_t ();
 
+        //  Interrupt blocking call if the app thread is stuck in one.
+        //  This function is is called from a different thread!
+        void stop ();
+
         //  Returns signaler associated with this application thread.
         struct i_signaler *get_signaler ();
 
         //  Processes commands sent to this thread (if any). If 'block' is
         //  set to true, returns only after at least one command was processed.
         //  If throttle argument is true, commands are processed at most once
-        //  in a predefined time period.
-        void process_commands (bool block_, bool throttle_);
+        //  in a predefined time period. The function returns false is the
+        //  associated context was terminated, true otherwise.
+        bool process_commands (bool block_, bool throttle_);
 
         //  Create a socket of a specified type.
         class socket_base_t *create_socket (int type_);
@@ -53,8 +58,14 @@ namespace zmq
         //  Unregister the socket from the app_thread (called by socket itself).
         void remove_socket (class socket_base_t *socket_);
 
+        //  Returns true is the associated context was already terminated.
+        bool is_terminated ();
+
     private:
 
+        //  Command handlers.
+        void process_stop ();
+
         //  All the sockets created from this application thread.
         typedef yarray_t <socket_base_t> sockets_t;
         sockets_t sockets;
@@ -65,6 +76,9 @@ namespace zmq
         //  Timestamp of when commands were processed the last time.
         uint64_t last_processing_time;
 
+        //  If true, 'stop' command was already received.
+        bool terminated;
+
         app_thread_t (const app_thread_t&);
         void operator = (const app_thread_t&);
     };
diff --git a/src/dispatcher.cpp b/src/dispatcher.cpp
index 93c5db3..a1154de 100644
--- a/src/dispatcher.cpp
+++ b/src/dispatcher.cpp
@@ -86,12 +86,19 @@ zmq::dispatcher_t::dispatcher_t (int app_threads_, int io_threads_,
 
 int zmq::dispatcher_t::term ()
 {
+    //  First send stop command to application threads so that any
+    //  blocking calls are interrupted.
+    for (app_threads_t::size_type i = 0; i != app_threads.size (); i++)
+        app_threads [i].app_thread->stop ();
+
+    //  Then mark context as terminated.
     term_sync.lock ();
     zmq_assert (!terminated);
     terminated = true;
     bool destroy = (sockets == 0);
     term_sync.unlock ();
     
+    //  If there are no sockets open, destroy the context immediately.
     if (destroy)
         delete this;
 
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 3b74359..b186683 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -37,6 +37,7 @@
 #include "platform.hpp"
 #include "pgm_sender.hpp"
 #include "pgm_receiver.hpp"
+#include "likely.hpp"
 
 zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
     object_t (parent_),
@@ -58,6 +59,11 @@ zmq::socket_base_t::~socket_base_t ()
 int zmq::socket_base_t::setsockopt (int option_, const void *optval_,
     size_t optvallen_)
 {
+    if (unlikely (app_thread->is_terminated ())) {
+        errno = ETERM;
+        return -1;
+    }
+
     //  First, check whether specific socket type overloads the option.
     int rc = xsetsockopt (option_, optval_, optvallen_);
     if (rc == 0 || errno != EINVAL)
@@ -71,6 +77,11 @@ int zmq::socket_base_t::setsockopt (int option_, const void *optval_,
 int zmq::socket_base_t::getsockopt (int option_, void *optval_,
     size_t *optvallen_)
 {
+    if (unlikely (app_thread->is_terminated ())) {
+        errno = ETERM;
+        return -1;
+    }
+
     if (option_ == ZMQ_RCVMORE) {
         if (*optvallen_ < sizeof (int64_t)) {
             errno = EINVAL;
@@ -86,6 +97,11 @@ int zmq::socket_base_t::getsockopt (int option_, void *optval_,
 
 int zmq::socket_base_t::bind (const char *addr_)
 {
+    if (unlikely (app_thread->is_terminated ())) {
+        errno = ETERM;
+        return -1;
+    }
+
     //  Parse addr_ string.
     std::string addr_type;
     std::string addr_args;
@@ -141,6 +157,11 @@ int zmq::socket_base_t::bind (const char *addr_)
 
 int zmq::socket_base_t::connect (const char *addr_)
 {
+    if (unlikely (app_thread->is_terminated ())) {
+        errno = ETERM;
+        return -1;
+    }
+
     //  Parse addr_ string.
     std::string addr_type;
     std::string addr_args;
@@ -328,13 +349,16 @@ int zmq::socket_base_t::connect (const char *addr_)
 
 int zmq::socket_base_t::send (::zmq_msg_t *msg_, int flags_)
 {
+    //  Process pending commands, if any.
+    if (unlikely (!app_thread->process_commands (false, true))) {
+        errno = ETERM;
+        return -1;
+    }
+
     //  At this point we impose the MORE flag on the message.
     if (flags_ & ZMQ_SNDMORE)
         msg_->flags |= ZMQ_MSG_MORE;
 
-    //  Process pending commands, if any.
-    app_thread->process_commands (false, true);
-
     //  Try to send the message.
     int rc = xsend (msg_, flags_);
     if (rc == 0)
@@ -350,7 +374,10 @@ int zmq::socket_base_t::send (::zmq_msg_t *msg_, int flags_)
     while (rc != 0) {
         if (errno != EAGAIN)
             return -1;
-        app_thread->process_commands (true, false);
+        if (unlikely (!app_thread->process_commands (true, false))) {
+            errno = ETERM;
+            return -1;
+        }
         rc = xsend (msg_, flags_);
     }
     return 0;
@@ -371,7 +398,10 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
     //  described above) from the one used by 'send'. This is because counting
     //  ticks is more efficient than doing rdtsc all the time.
     if (++ticks == inbound_poll_rate) {
-        app_thread->process_commands (false, false);
+        if (unlikely (!app_thread->process_commands (false, false))) {
+            errno = ETERM;
+            return -1;
+        }
         ticks = 0;
     }
 
@@ -392,7 +422,10 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
     if (flags_ & ZMQ_NOBLOCK) {
         if (errno != EAGAIN)
             return -1;
-        app_thread->process_commands (false, false);
+        if (unlikely (!app_thread->process_commands (false, false))) {
+            errno = ETERM;
+            return -1;
+        }
         ticks = 0;
         return xrecv (msg_, flags_);
     }
@@ -402,7 +435,10 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
     while (rc != 0) {
         if (errno != EAGAIN)
             return -1;
-        app_thread->process_commands (true, false);
+        if (unlikely (!app_thread->process_commands (true, false))) {
+            errno = ETERM;
+            return -1;
+        }
         rc = xrecv (msg_, flags_);
         ticks = 0;
     }
diff --git a/src/zmq.cpp b/src/zmq.cpp
index b8d18f7..0681cbb 100644
--- a/src/zmq.cpp
+++ b/src/zmq.cpp
@@ -88,6 +88,8 @@ const char *zmq_strerror (int errnum_)
         return "Operation cannot be accomplished in current state";
     case ENOCOMPATPROTO:
         return "The protocol is not compatible with the socket type";
+    case ETERM:
+        return "Context was terminated";
     default:
 #if defined _MSC_VER
 #pragma warning (push)