msg/Pipe: release dispatch throttle on delayed queue discard

This avoids leaking into the throttle and deadlocking.

Signed-off-by: Sage Weil <sage@inktank.com>

2 files changed, 12 insertions, 8 deletions

diff --git a/src/msg/Pipe.cc b/src/msg/Pipe.cc
index f6061564444..2bbaf0bc17f 100644
--- a/src/msg/Pipe.cc
+++ b/src/msg/Pipe.cc
@@ -158,6 +158,17 @@ void Pipe::join_reader()
   reader_needs_join = false;
 }
 
+void Pipe::DelayedDelivery::discard()
+{
+  Mutex::Locker l(delay_lock);
+  while (!delay_queue.empty()) {
+    Message *m = delay_queue.front().second;
+    pipe->msgr->dispatch_throttle_release(m->get_dispatch_throttle_size());
+    m->put();
+    delay_queue.pop_front();
+  }
+}
+
 void *Pipe::DelayedDelivery::entry()
 {
   Mutex::Locker locker(delay_lock);
@@ -1053,7 +1064,6 @@ void Pipe::discard_out_queue()
   out_q.clear();
 }
 
-
 void Pipe::fault(bool onread)
 {
   const md_config_t *conf = msgr->cct->_conf;
diff --git a/src/msg/Pipe.h b/src/msg/Pipe.h
index 4754a530585..30fb7eede11 100644
--- a/src/msg/Pipe.h
+++ b/src/msg/Pipe.h
@@ -89,13 +89,7 @@ class DispatchQueue;
 	delay_queue.push_back(make_pair(release, m));
 	delay_cond.Signal();
       }
-      void discard() {
-	Mutex::Locker l(delay_lock);
-	while (!delay_queue.empty()) {
-	  delay_queue.front().second->put();
-	  delay_queue.pop_front();
-	}
-      }
+      void discard();
       void stop() {
 	delay_lock.Lock();
 	stop_delayed_delivery = true;