workqueue: drain

diff --git a/src/common/WorkQueue.cc b/src/common/WorkQueue.cc
index 5ce590b77a3a71292744bb13852361f19bed035d..fc315c34ada0750baf1a5d6b0beec05295060081 100644 (file)
--- a/src/common/WorkQueue.cc
+++ b/src/common/WorkQueue.cc
@@ -44,7 +44,7 @@ void ThreadPool::worker()
          _lock.Lock();
          dout(15) << "worker wq " << wq->name << " done processing " << item << dendl;
          processing--;
-         if (_pause)
+         if (_pause || _draining)
            _wait_cond.Signal();
          did = true;
          break;
@@ -118,3 +118,15 @@ void ThreadPool::unpause()
   _cond.Signal();
   _lock.Unlock();
 }
+
+void ThreadPool::drain()
+{
+  dout(10) << "drain" << dendl;
+  _lock.Lock();
+  _draining = true;
+  while (processing)
+    _wait_cond.Wait(_lock);
+  _draining = false;
+  _lock.Unlock();
+}
+

diff --git a/src/common/WorkQueue.h b/src/common/WorkQueue.h
index 86b61c6879c29a0316af9c5234aad4d4129c33e9..fcab0d8de41a0ad27dde4411f4e9d209c1961a37 100644 (file)
--- a/src/common/WorkQueue.h
+++ b/src/common/WorkQueue.h
@@ -23,7 +23,7 @@ class ThreadPool {
   string name;
   Mutex _lock;
   Cond _cond;
-  bool _stop, _pause;
+  bool _stop, _pause, _draining;
   Cond _wait_cond;
 
   struct _WorkQueue {
@@ -116,7 +116,7 @@ public:
     name(nm),
     _lock((new string(name + "::lock"))->c_str()),  // deliberately leak this
     _stop(false),
-    _pause(false),
+    _pause(false), _draining(false),
     last_work_queue(0),
     processing(0) {
     set_num_threads(n);
@@ -169,6 +169,7 @@ public:
   void pause();
   void pause_new();
   void unpause();
+  void drain();
 };