From bb13e5e60b83bc44d436d4fdf41cb56af4da7c81 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Fri, 8 Jun 2012 14:32:15 +0000 Subject: QPID-4046: rate limit the release of dequeued messages. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1348090 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qpid/broker/MessageDeque.cpp | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) (limited to 'qpid/cpp/src') diff --git a/qpid/cpp/src/qpid/broker/MessageDeque.cpp b/qpid/cpp/src/qpid/broker/MessageDeque.cpp index 474e4139bd..f26d5841a6 100644 --- a/qpid/cpp/src/qpid/broker/MessageDeque.cpp +++ b/qpid/cpp/src/qpid/broker/MessageDeque.cpp @@ -40,13 +40,16 @@ size_t MessageDeque::index(const framing::SequenceNumber& position) bool MessageDeque::deleted(const QueuedMessage& m) { size_t i = index(m.position); - if (i < messages.size() && messages[i].status != QueuedMessage::DELETED) { - messages[i].status = QueuedMessage::DELETED; - clean(); - return true; - } else { - return false; + if (i < messages.size()) { + QueuedMessage *qm = &messages[i]; + if (qm->status != QueuedMessage::DELETED) { + qm->status = QueuedMessage::DELETED; + qm->payload.reset(); // message no longer needed + clean(); + return true; + } } + return false; } size_t MessageDeque::size() @@ -144,6 +147,7 @@ QueuedMessage* MessageDeque::pushPtr(const QueuedMessage& added) { messages.back().status = QueuedMessage::AVAILABLE; if (head >= messages.size()) head = messages.size() - 1; ++available; + clean(); // QPID-4046: let producer help clean the backlog of deleted messages return &messages.back(); } @@ -195,10 +199,15 @@ void MessageDeque::setPosition(const framing::SequenceNumber& n) { void MessageDeque::clean() { - while (messages.size() && messages.front().status == QueuedMessage::DELETED) { + // QPID-4046: If a queue has multiple consumers, then it is possible for a large + // collection of deleted messages to build up. Limit the number of messages cleaned + // up on each call to clean(). + size_t count = 0; + while (messages.size() && messages.front().status == QueuedMessage::DELETED && count < 10) { messages.pop_front(); - if (head) --head; + count += 1; } + head = (head > count) ? head - count : 0; } void MessageDeque::foreach(Functor f) -- cgit v1.2.1