Updated AutoDelete, now only run when auto delete queues exist and

additional auto delete queues being declared. cuts 1-3% of CPU time 
for concurrent publisher run.

Could do with some more auto delete tests...




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

5 files changed, 26 insertions, 44 deletions

diff --git a/qpid/cpp/src/qpid/broker/AutoDelete.cpp b/qpid/cpp/src/qpid/broker/AutoDelete.cpp
index 45ce12c036..6fa3bfbebf 100644
--- a/qpid/cpp/src/qpid/broker/AutoDelete.cpp
+++ b/qpid/cpp/src/qpid/broker/AutoDelete.cpp
@@ -25,12 +25,12 @@
 using namespace qpid::broker;
 using namespace qpid::sys;
 
-AutoDelete::AutoDelete(QueueRegistry* const _registry, uint32_t _period)
-    : registry(_registry), period(_period*TIME_MSEC), stopped(true) { }
+AutoDelete::AutoDelete(QueueRegistry* const _registry, uint32_t _water_mark)
+    : registry(_registry), high_water_mark(_water_mark), water_mark(0) { }
 
 void AutoDelete::add(Queue::shared_ptr const queue){
-    Mutex::ScopedLock l(lock);
-    queues.push(queue);
+     	Mutex::ScopedLock l(lock);
+    	queues.push(queue);
 }
 
 Queue::shared_ptr const AutoDelete::pop(){
@@ -43,7 +43,15 @@ Queue::shared_ptr const AutoDelete::pop(){
     return next;
 }
 
-void AutoDelete::process(){
+void AutoDelete::clean(){
+    if (water_mark++ < high_water_mark)
+    	return;
+    water_mark =0;
+    cleanNow();
+}
+
+
+void AutoDelete::cleanNow(){
     Queue::shared_ptr seen;
     for(Queue::shared_ptr q = pop(); q; q = pop()){
         if(seen == q){
@@ -58,30 +66,9 @@ void AutoDelete::process(){
             if(!seen) seen = q;
         }
     }
+    
 }
 
-void AutoDelete::run(){
-    Monitor::ScopedLock l(monitor);
-    while(!stopped){
-        process();
-        monitor.wait(AbsTime(now(), period));
-    }
-}
 
-void AutoDelete::start(){
-    Monitor::ScopedLock l(monitor);
-    if(stopped){
-        stopped = false;
-        runner = Thread(this);
-    }
-}
 
-void AutoDelete::stop(){
-    {
-        Monitor::ScopedLock l(monitor);
-        if(stopped) return;
-        stopped = true;
-    }
-    monitor.notify();
-    runner.join();
-}
+
diff --git a/qpid/cpp/src/qpid/broker/AutoDelete.h b/qpid/cpp/src/qpid/broker/AutoDelete.h
index d4d0f84f4b..c4d46ca505 100644
--- a/qpid/cpp/src/qpid/broker/AutoDelete.h
+++ b/qpid/cpp/src/qpid/broker/AutoDelete.h
@@ -30,24 +30,20 @@
 
 namespace qpid {
     namespace broker{
-        class AutoDelete : private qpid::sys::Runnable {
+        class AutoDelete {
             qpid::sys::Mutex lock;            
-            qpid::sys::Monitor monitor;            
             std::queue<Queue::shared_ptr> queues;
             QueueRegistry* const registry;
-            sys::Duration period;
-            volatile bool stopped;
-            qpid::sys::Thread runner;
-            
+            uint32_t high_water_mark;
+	    uint32_t water_mark;
+             
             Queue::shared_ptr const pop();
-            void process();
-            virtual void run();
 
         public:
-            AutoDelete(QueueRegistry* const registry, uint32_t period);
+            AutoDelete(QueueRegistry* const registry, uint32_t _water_mark);
             void add(Queue::shared_ptr const);
-            void start();
-            void stop();
+            void clean();
+	    void cleanNow();
         };
     }
 }
diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 65252f5415..7c96322f02 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -82,7 +82,7 @@ Broker::Broker(const Broker::Options& conf) :
     queues(store.get()),
     timeout(30000),
     stagingThreshold(0),
-    cleaner(&queues, timeout/10),
+    cleaner(&queues, 100),  // clean every 100 auto delete declares.
     factory(*this),
     dtxManager(store.get())
 {
@@ -98,7 +98,6 @@ Broker::Broker(const Broker::Options& conf) :
         store->recover(recoverer);
     }
 
-    cleaner.start();
 }
 
 
@@ -128,7 +127,7 @@ void Broker::run() {
 void Broker::shutdown() {
     if (acceptor)
         acceptor->shutdown();
-    cleaner.stop();
+//cct    cleaner.cleanNow();  // do we need to delete on close?
 }
 
 Broker::~Broker() {
diff --git a/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp b/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
index b7a61ababe..a0ab42592e 100644
--- a/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
+++ b/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
@@ -219,7 +219,6 @@ void BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, uint
 	assert(queue);
 	if (queue_created.second) { // This is a new queue
 	    channel.setDefaultQueue(queue);
-
             //apply settings & create persistent record if required
             queue_created.first->create(arguments);
 
@@ -232,6 +231,7 @@ void BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, uint
 		connection.exclusiveQueues.push_back(queue);
 	    } else if(autoDelete){
 		broker.getCleaner().add(queue);
+		broker.getCleaner().clean(); // check if cleaning is needed
 	    }
 	}
     }
diff --git a/qpid/cpp/src/qpid/broker/ConnectionFactory.cpp b/qpid/cpp/src/qpid/broker/ConnectionFactory.cpp
index 5372bfe377..4429a2583c 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionFactory.cpp
+++ b/qpid/cpp/src/qpid/broker/ConnectionFactory.cpp
@@ -31,7 +31,7 @@ ConnectionFactory::ConnectionFactory(Broker& b) : broker(b)
 
 ConnectionFactory::~ConnectionFactory()
 {
-    broker.getCleaner().stop();
+
 }
 
 qpid::sys::ConnectionInputHandler*