msg: QueueStrategy::wait() joins all threads

wait() was only looping over disp_threads, which is an intrusive list
that only contains threads that are waiting on a message from
ds_dispatch(). this means that some QSThreads could outlive the
QueueStrategy itself, causing a segfault in QueueStrategy::entry()

Fixes: http://tracker.ceph.com/issues/20534
Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/msg/QueueStrategy.cc b/src/msg/QueueStrategy.cc
index 0ce279b31e9f1055abd6b79545a53df6d76a840e..e41ab79fdeb34671ea4a954d8a4f4d3022030dac 100644 (file)
--- a/src/msg/QueueStrategy.cc
+++ b/src/msg/QueueStrategy.cc
@@ -15,6 +15,7 @@
 #include "QueueStrategy.h"
 #define dout_subsys ceph_subsys_ms
 #include "common/debug.h"
+#include "common/backport14.h"
 
 QueueStrategy::QueueStrategy(int _n_threads)
   : lock("QueueStrategy::lock"),
@@ -85,16 +86,13 @@ void QueueStrategy::shutdown()
 
 void QueueStrategy::wait()
 {
-  QSThread *thrd;
   lock.Lock();
   assert(stop);
-  while (disp_threads.size()) {
-    thrd = &(disp_threads.front());
-    disp_threads.pop_front();
+  for (auto& thread : threads) {
     lock.Unlock();
 
     // join outside of lock
-    thrd->join();
+    thread->join();
 
     lock.Lock();
   }
@@ -103,14 +101,15 @@ void QueueStrategy::wait()
 
 void QueueStrategy::start()
 {
-  QSThread *thrd;
   assert(!stop);
   lock.Lock();
+  threads.reserve(n_threads);
   for (int ix = 0; ix < n_threads; ++ix) {
     string thread_name = "ms_xio_qs_";
     thread_name.append(std::to_string(ix));
-    thrd = new QSThread(this);
+    auto thrd = ceph::make_unique<QSThread>(this);
     thrd->create(thread_name.c_str());
+    threads.emplace_back(std::move(thrd));
   }
   lock.Unlock();
 }

diff --git a/src/msg/QueueStrategy.h b/src/msg/QueueStrategy.h
index 41f28bb9e7105dc27b584d86c6e0ac2f2f76fe17..a531cd7774327a601478ff77f725a33199af6fd4 100644 (file)
--- a/src/msg/QueueStrategy.h
+++ b/src/msg/QueueStrategy.h
@@ -16,6 +16,8 @@
 #ifndef QUEUE_STRATEGY_H
 #define QUEUE_STRATEGY_H
 
+#include <vector>
+#include <memory>
 #include <boost/intrusive/list.hpp>
 #include "DispatchStrategy.h"
 #include "msg/Messenger.h"
@@ -24,7 +26,7 @@ namespace bi = boost::intrusive;
 
 class QueueStrategy : public DispatchStrategy {
   Mutex lock;
-  int n_threads;
+  const int n_threads;
   bool stop;
 
   Message::Queue mqueue;
@@ -37,7 +39,6 @@ class QueueStrategy : public DispatchStrategy {
     explicit QSThread(QueueStrategy *dq) : thread_q(), dq(dq), cond() {}
     void* entry() {
       dq->entry(this);
-      delete(this);
       return NULL;
     }
 
@@ -47,7 +48,8 @@ class QueueStrategy : public DispatchStrategy {
                                       &QSThread::thread_q > > Queue;
   };
 
-  QSThread::Queue disp_threads;
+  std::vector<std::unique_ptr<QSThread>> threads; //< all threads
+  QSThread::Queue disp_threads; //< waiting threads
 
 public:
   explicit QueueStrategy(int n_threads);