summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/msg/Pipe.cc12
-rw-r--r--src/msg/Pipe.h8
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;