Add blocking to broker::Timer/TimerTask so cancel and fire()

cannot be interleaved.


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

2 files changed, 15 insertions, 3 deletions

diff --git a/qpid/cpp/src/qpid/broker/Timer.cpp b/qpid/cpp/src/qpid/broker/Timer.cpp
index 53a1597deb..7832204e2a 100644
--- a/qpid/cpp/src/qpid/broker/Timer.cpp
+++ b/qpid/cpp/src/qpid/broker/Timer.cpp
@@ -26,6 +26,8 @@ using qpid::sys::AbsTime;
 using qpid::sys::Duration;
 using qpid::sys::Monitor;
 using qpid::sys::Thread;
+using qpid::sys::Mutex;
+using qpid::sys::ScopedLock;
 using namespace qpid::broker;
 
 TimerTask::TimerTask(Duration timeout) :
@@ -38,7 +40,11 @@ TimerTask::~TimerTask(){}
 
 void TimerTask::reset() { time = AbsTime(AbsTime::now(), duration); }
 
-void TimerTask::cancel() { cancelled = true; }
+void TimerTask::cancel() {
+    ScopedLock<Mutex> l(cancelLock);
+    cancelled = true;
+}
+
 bool TimerTask::isCancelled() const { return cancelled; }
 
 Timer::Timer() : active(false)
@@ -60,16 +66,21 @@ void Timer::run()
         } else {
             intrusive_ptr<TimerTask> t = tasks.top();
             tasks.pop();
+            {
+            ScopedLock<Mutex> l(t->cancelLock);
             if (t->isCancelled()) {
+                continue;
             } else if(t->time < AbsTime::now()) {
                 Monitor::ScopedUnlock u(monitor);
                 t->fire();
+                continue;
             } else {
                 // If the timer was adjusted into the future it might no longer
                 // be the next event, so push and then get top to make sure
                 tasks.push(t);
-                monitor.wait(tasks.top()->time);
             }
+            }
+            monitor.wait(tasks.top()->time);
         }
     }
 }
diff --git a/qpid/cpp/src/qpid/broker/Timer.h b/qpid/cpp/src/qpid/broker/Timer.h
index 564fec5804..7218fb0e79 100644
--- a/qpid/cpp/src/qpid/broker/Timer.h
+++ b/qpid/cpp/src/qpid/broker/Timer.h
@@ -43,6 +43,7 @@ struct TimerTask : public RefCounted {
     const qpid::sys::Duration duration;
     qpid::sys::AbsTime time;
     volatile bool cancelled;
+    qpid::sys::Mutex cancelLock;
 
     QPID_BROKER_EXTERN TimerTask(qpid::sys::Duration timeout);
     TimerTask(qpid::sys::AbsTime time);
@@ -60,7 +61,7 @@ struct Later {
 
 class Timer : private qpid::sys::Runnable {
   protected:
-    qpid::sys::Monitor monitor;            
+    qpid::sys::Monitor monitor;
     std::priority_queue<boost::intrusive_ptr<TimerTask>,
                         std::vector<boost::intrusive_ptr<TimerTask> >,
                         Later> tasks;