QPID-3280: Performance problem with TTL messages.

When sending a large number of messages with nonzero TTLs to a
cluster, overall message throughput drops by around 20-30% compared to
messages with TTL 0.

The previous approach to TTL in the cluster is replaced with a simpler
"cluster clock". Also QueueCleaner is executed in the cluster timer,
and modified to be deterministic in a cluster.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1136170 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 12 insertions, 6 deletions

diff --git a/cpp/src/qpid/broker/QueueCleaner.cpp b/cpp/src/qpid/broker/QueueCleaner.cpp
index 3499ea8a4d..838bc28be8 100644
--- a/cpp/src/qpid/broker/QueueCleaner.cpp
+++ b/cpp/src/qpid/broker/QueueCleaner.cpp
@@ -7,9 +7,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- * 
+ *
  *   http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -27,7 +27,7 @@
 namespace qpid {
 namespace broker {
 
-QueueCleaner::QueueCleaner(QueueRegistry& q, sys::Timer& t) : queues(q), timer(t) {}
+QueueCleaner::QueueCleaner(QueueRegistry& q, sys::Timer* t) : queues(q), timer(t) {}
 
 QueueCleaner::~QueueCleaner()
 {
@@ -36,10 +36,16 @@ QueueCleaner::~QueueCleaner()
 
 void QueueCleaner::start(qpid::sys::Duration p)
 {
+    period = p;
     task = new Task(*this, p);
-    timer.add(task);
+    timer->add(task);
 }
 
+void QueueCleaner::setTimer(qpid::sys::Timer* timer) {
+    this->timer = timer;
+}
+
+
 QueueCleaner::Task::Task(QueueCleaner& p, qpid::sys::Duration d) : sys::TimerTask(d,"QueueCleaner"), parent(p) {}
 
 void QueueCleaner::Task::fire()
@@ -65,9 +71,9 @@ void QueueCleaner::fired()
     std::vector<Queue::shared_ptr> copy;
     CollectQueues collect(&copy);
     queues.eachQueue(collect);
-    std::for_each(copy.begin(), copy.end(), boost::bind(&Queue::purgeExpired, _1));
+    std::for_each(copy.begin(), copy.end(), boost::bind(&Queue::purgeExpired, _1, period));
     task->setupNextFire();
-    timer.add(task);
+    timer->add(task);
 }