qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r1535803 - in /qpid/trunk/qpid/cpp/src/qpid/broker: LossyQueue.cpp Queue.cpp Queue.h
Date Fri, 25 Oct 2013 18:13:35 GMT
Author: aconway
Date: Fri Oct 25 18:13:35 2013
New Revision: 1535803

URL: http://svn.apache.org/r1535803
Log:
QPID-4287: Poor performance when a priority queue with a ring queue policy has a large backlog

LossyQueue::checkDepth was performing an unintended linear search of its
messages when a new message was added at capacity.  Since the messages are in
priority order, only the tail message on the queue needs to be compared with the
new message to determine which of them should be dropped.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp?rev=1535803&r1=1535802&r2=1535803&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp Fri Oct 25 18:13:35 2013
@@ -51,8 +51,19 @@ bool LossyQueue::checkDepth(const QueueD
     while (settings.maxDepth && (settings.maxDepth - current < increment)) {
         QPID_LOG(debug, "purging " << name << ": current depth is [" <<
current << "], max depth is [" << settings.maxDepth << "], new message has
size " << increment.getSize());
         qpid::sys::Mutex::ScopedUnlock u(messageLock);
-        //TODO: arguably we should try and purge expired messages first but that is potentially
expensive
-        if (remove(1, settings.priorities ? boost::bind(&isLowerPriorityThan, message.getPriority(),
_1) : MessagePredicate(), boost::bind(&reroute, alternateExchange, _1), PURGE, false))
{
+        //TODO: arguably we should try and purge expired messages first but that
+        //is potentially expensive
+
+        // Note: in the case of a priority queue we are only comparing the new mesage
+        // with single lowest-priority message, hence the final parameter maxTests
+        // is 1 in this case, so we only test one message for removal.
+        if (remove(1,
+                   settings.priorities ?
+                   boost::bind(&isLowerPriorityThan, message.getPriority(), _1) :
+                   MessagePredicate(), boost::bind(&reroute, alternateExchange, _1),
+                   PURGE, false,
+                   settings.priorities ? 1 : 0))
+        {
             if (mgmtObject) {
                 mgmtObject->inc_discardsRing(1);
                 if (brokerMgmtObject)

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1535803&r1=1535802&r2=1535803&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Fri Oct 25 18:13:35 2013
@@ -706,30 +706,29 @@ namespace {
     }
 } // end namespace
 
-uint32_t Queue::remove(const uint32_t maxCount, MessagePredicate p, MessageFunctor f, SubscriptionType
type, bool triggerAutoDelete)
+uint32_t Queue::remove(const uint32_t maxCount, MessagePredicate p, MessageFunctor f,
+                       SubscriptionType type, bool triggerAutoDelete, uint32_t maxTests)
 {
     ScopedAutoDelete autodelete(*this);
     std::deque<Message> removed;
     {
         QueueCursor c(type);
-        uint32_t count(0);
+        uint32_t count(0), tests(0);
         Mutex::ScopedLock locker(messageLock);
         Message* m = messages->next(c);
         while (m){
+            if (maxTests && tests++ >= maxTests) break;
             if (!p || p(*m)) {
-                if (!maxCount || count++ < maxCount) {
-                    if (m->getState() == AVAILABLE) {
-                        //don't actually acquire, just act as if we did
-                        observeAcquire(*m, locker);
-                    }
-                    observeDequeue(*m, locker, triggerAutoDelete ? &autodelete : 0);
-                    removed.push_back(*m);//takes a copy of the message
-                    if (!messages->deleted(c)) {
-                        QPID_LOG(warning, "Failed to correctly remove message from " <<
name << "; state is not consistent!");
-                        assert(false);
-                    }
-                } else {
-                    break;
+                if (maxCount && count++ >= maxCount) break;
+                if (m->getState() == AVAILABLE) {
+                    //don't actually acquire, just act as if we did
+                    observeAcquire(*m, locker);
+                }
+                observeDequeue(*m, locker, triggerAutoDelete ? &autodelete : 0);
+                removed.push_back(*m);//takes a copy of the message
+                if (!messages->deleted(c)) {
+                    QPID_LOG(warning, "Failed to correctly remove message from " <<
name << "; state is not consistent!");
+                    assert(false);
                 }
             }
             m = messages->next(c);

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=1535803&r1=1535802&r2=1535803&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Fri Oct 25 18:13:35 2013
@@ -255,7 +255,22 @@ class Queue : public boost::enable_share
     void abandoned(const Message& message);
     bool checkNotDeleted(const Consumer::shared_ptr&);
     void notifyDeleted();
-    uint32_t remove(uint32_t maxCount, MessagePredicate, MessageFunctor, SubscriptionType,
bool triggerAutoDelete);
+
+    /** Remove messages from the queue:
+     *@param maxCount Maximum number of messages to remove, 0 means unlimited.
+     *@param p Only remove messages for which p(msg) is true.
+     *@param f Call f on each message that is removed.
+     *@param st Use a cursor of this SubscriptionType to iterate messages to remove.
+     *@param triggerAutoDelete If true removing messages may trigger aut-delete.
+     *@param maxTests Max number of messages to test for removal, 0 means unlimited.
+     *@return Number of messages removed.
+     */
+    uint32_t remove(uint32_t maxCount,
+                    MessagePredicate p, MessageFunctor f,
+                    SubscriptionType st,
+                    bool triggerAutoDelete,
+                    uint32_t maxTests=0);
+
     virtual bool checkDepth(const QueueDepth& increment, const Message&);
     void tryAutoDelete();
   public:



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message