From ea6880f8a2d12d7de62c61feae9e55cf5d10c7af Mon Sep 17 00:00:00 2001 From: Sage Weil Date: Tue, 11 Jun 2013 19:27:01 -0700 Subject: [PATCH] msg/DispatchQueue: do not discard queued events on stop When the shutdown/stop flag is set, continue to work through the queue. Process events, but discard messages. This avoids the loss of reset events on shutdown that are necessary to clean up ref cycles. Signed-off-by: Sage Weil --- src/msg/DispatchQueue.cc | 50 +++++++++++++++++++++++----------------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/src/msg/DispatchQueue.cc b/src/msg/DispatchQueue.cc index a92c357beb55..3adc038e87f9 100644 --- a/src/msg/DispatchQueue.cc +++ b/src/msg/DispatchQueue.cc @@ -78,8 +78,8 @@ void DispatchQueue::local_delivery(Message *m, int priority) void DispatchQueue::entry() { lock.Lock(); - while (!stop) { - while (!mqueue.empty() && !stop) { + while (true) { + while (!mqueue.empty()) { QueueItem qitem = mqueue.dequeue(); if (!qitem.is_code()) remove_arrival(qitem.get_message()); @@ -104,29 +104,37 @@ void DispatchQueue::entry() } } else { Message *m = qitem.get_message(); - uint64_t msize = m->get_dispatch_throttle_size(); - m->set_dispatch_throttle_size(0); // clear it out, in case we requeue this message. - - ldout(cct,1) << "<== " << m->get_source_inst() - << " " << m->get_seq() - << " ==== " << *m - << " ==== " << m->get_payload().length() << "+" << m->get_middle().length() - << "+" << m->get_data().length() - << " (" << m->get_footer().front_crc << " " << m->get_footer().middle_crc - << " " << m->get_footer().data_crc << ")" - << " " << m << " con " << m->get_connection() - << dendl; - msgr->ms_deliver_dispatch(m); - - msgr->dispatch_throttle_release(msize); - - ldout(cct,20) << "done calling dispatch on " << m << dendl; + if (stop) { + ldout(cct,10) << " stop flag set, discarding " << m << " " << *m << dendl; + m->put(); + } else { + uint64_t msize = m->get_dispatch_throttle_size(); + m->set_dispatch_throttle_size(0); // clear it out, in case we requeue this message. + + ldout(cct,1) << "<== " << m->get_source_inst() + << " " << m->get_seq() + << " ==== " << *m + << " ==== " << m->get_payload().length() << "+" << m->get_middle().length() + << "+" << m->get_data().length() + << " (" << m->get_footer().front_crc << " " << m->get_footer().middle_crc + << " " << m->get_footer().data_crc << ")" + << " " << m << " con " << m->get_connection() + << dendl; + msgr->ms_deliver_dispatch(m); + + msgr->dispatch_throttle_release(msize); + + ldout(cct,20) << "done calling dispatch on " << m << dendl; + } } lock.Lock(); } - if (!stop) - cond.Wait(lock); //wait for something to be put on queue + if (stop) + break; + + // wait for something to be put on queue + cond.Wait(lock); } lock.Unlock(); } -- 2.47.3