[zeromq-dev] [PATCH] Fix potential race when terminating io threads in ctx_t destructor.

[Martin Townsend]

After stopping the IO threads wait for the associated poller to stop 
before destroying them to avoid get_load () == 0 assertion.
Potentially a fix for Issue #795

---
  src/ctx.cpp       | 12 ++++++++++--
  src/epoll.cpp     | 20 +++++++++++++++++++-
  src/epoll.hpp     |  4 ++++
  src/io_thread.cpp |  5 +++++
  src/io_thread.hpp |  3 +++
  5 files changed, 41 insertions(+), 3 deletions(-)

diff --git a/src/ctx.cpp b/src/ctx.cpp
index 0155ef2..72db140 100644
--- a/src/ctx.cpp
+++ b/src/ctx.cpp
@@ -80,8 +80,16 @@ zmq::ctx_t::~ctx_t ()
          io_threads [i]->stop ();

      //  Wait till I/O threads actually terminate.
-    for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
-        delete io_threads [i];
+    for (io_threads_t::size_type i = 0; i != io_threads.size (); i++) {
+        int j = 0;
+
+        while (!io_threads [i]->stopped ()) {
+            sleep (1);
+            if (++j == 5)
+                break;
+        }
+        delete io_threads [i];
+    }

      //  Deallocate the reaper thread object.
      delete reaper;
diff --git a/src/epoll.cpp b/src/epoll.cpp
index 324d4bd..2c3a1b9 100644
--- a/src/epoll.cpp
+++ b/src/epoll.cpp
@@ -33,7 +33,8 @@
  #include "i_poll_events.hpp"

  zmq::epoll_t::epoll_t () :
-    stopping (false)
+    stopping (false),
+    poller_running (false)
  {
      epoll_fd = epoll_create (1);
      errno_assert (epoll_fd != -1);
@@ -41,12 +42,22 @@ zmq::epoll_t::epoll_t () :

  zmq::epoll_t::~epoll_t ()
  {
+    int i;
+
      //  Wait till the worker thread exits.
      worker.stop ();

      close (epoll_fd);
      for (retired_t::iterator it = retired.begin (); it != retired.end 
(); ++it)
          delete *it;
+
+    // Allow 5 seconds for the poller worker thread to exit.
+    i = 0;
+    while (poller_running) {
+        sleep (1);
+        if (++i == 5)
+            break;
+    }
  }

  zmq::epoll_t::handle_t zmq::epoll_t::add_fd (fd_t fd_, i_poll_events 
*events_)
@@ -126,6 +137,11 @@ void zmq::epoll_t::stop ()
      stopping = true;
  }

+bool zmq::epoll_t::stopped ()
+{
+    return !poller_running;
+}
+
  int zmq::epoll_t::max_fds ()
  {
      return -1;
@@ -135,6 +151,7 @@ void zmq::epoll_t::loop ()
  {
      epoll_event ev_buf [max_io_events];

+    poller_running = true;
      while (!stopping) {

          //  Execute any due timers.
@@ -171,6 +188,7 @@ void zmq::epoll_t::loop ()
              delete *it;
          retired.clear ();
      }
+    poller_running = false;
  }

  void zmq::epoll_t::worker_routine (void *arg_)
diff --git a/src/epoll.hpp b/src/epoll.hpp
index fbf871c..4a29c48 100644
--- a/src/epoll.hpp
+++ b/src/epoll.hpp
@@ -57,6 +57,7 @@ namespace zmq
          void reset_pollout (handle_t handle_);
          void start ();
          void stop ();
+        bool stopped ();

          static int max_fds ();

@@ -85,6 +86,9 @@ namespace zmq
          //  If true, thread is in the process of shutting down.
          bool stopping;

+        //  If true, poller worker thread is running.
+        bool poller_running;
+
          //  Handle of the physical thread doing the I/O work.
          thread_t worker;

diff --git a/src/io_thread.cpp b/src/io_thread.cpp
index b05b56b..59295a0 100644
--- a/src/io_thread.cpp
+++ b/src/io_thread.cpp
@@ -50,6 +50,11 @@ void zmq::io_thread_t::stop ()
      send_stop ();
  }

+bool zmq::io_thread_t::stopped ()
+{
+    return poller->stopped ();
+}
+
  zmq::mailbox_t *zmq::io_thread_t::get_mailbox ()
  {
      return &mailbox;
diff --git a/src/io_thread.hpp b/src/io_thread.hpp
index 49a2e80..e4c7797 100644
--- a/src/io_thread.hpp
+++ b/src/io_thread.hpp
@@ -52,6 +52,9 @@ namespace zmq
          //  Ask underlying thread to stop.
          void stop ();

+        //  Has it stopped, used after calling stop above.
+        bool stopped ();
+
          //  Returns mailbox associated with this I/O thread.
          mailbox_t *get_mailbox ();

-- 
1.9.1