support cancelling timers.

Author: chenshuo

ChangeLog

@@ -1,3 +1,10 @@
+2011-09-18  Shuo Chen  <[email protected]>
+
+  * EventLoop now supports cancelling timer.
+  * Add two examples of asio chat server, demo copy-on-write
+  in mulrithreaded program.
+  * Version 0.2.9
+
 2011-09-04  Shuo Chen  <[email protected]>
 
   * Refactored RPC implementation of version 1 and 2,

muduo/net/EventLoop.cc

@@ -172,6 +172,11 @@ TimerId EventLoop::runEvery(double interval, const TimerCallback& cb)
   return timerQueue_->addTimer(cb, time, interval);
 }
 
+void EventLoop::cancel(TimerId timerId)
+{
+  return timerQueue_->cancel(timerId);
+}
+
 void EventLoop::updateChannel(Channel* channel)
 {
   assert(channel->ownerLoop() == this);

muduo/net/EventLoop.h

@@ -89,7 +89,7 @@ class EventLoop : boost::noncopyable
   /// Cancels the timer.
   /// Safe to call from other threads.
   ///
-  // void cancel(TimerId timerId);
+  void cancel(TimerId timerId);
 
   // internal usage
   void wakeup();

muduo/net/TimerId.h

@@ -27,15 +27,17 @@ class TimerId : public muduo::copyable
 {
  public:
   TimerId(Timer* timer, int64_t seq)
-    : value_(timer),
+    : timer_(timer),
       seq_(seq)
   {
   }
 
   // default copy-ctor, dtor and assignment are okay
 
+  friend class TimerQueue;
+
  private:
-  Timer* value_;
+  Timer* timer_;
   int64_t seq_;
 };
 

muduo/net/TimerQueue.cc

@@ -90,7 +90,8 @@ TimerQueue::TimerQueue(EventLoop* loop)
   : loop_(loop),
     timerfd_(createTimerfd()),
     timerfdChannel_(loop, timerfd_),
-    timers_()
+    timers_(),
+    callingExpiredTimers_(false)
 {
   timerfdChannel_.setReadCallback(
       boost::bind(&TimerQueue::handleRead, this));
@@ -119,6 +120,12 @@ TimerId TimerQueue::addTimer(const TimerCallback& cb,
   return TimerId(timer, timer->sequence());
 }
 
+void TimerQueue::cancel(TimerId timerId)
+{
+  loop_->runInLoop(
+      boost::bind(&TimerQueue::cancelInLoop, this, timerId));
+}
+
 void TimerQueue::scheduleInLoop(Timer* timer)
 {
   loop_->assertInLoopThread();
@@ -130,6 +137,26 @@ void TimerQueue::scheduleInLoop(Timer* timer)
   }
 }
 
+void TimerQueue::cancelInLoop(TimerId timerId)
+{
+  loop_->assertInLoopThread();
+  assert(timers_.size() == activeTimers_.size());
+  ActiveTimer timer(timerId.timer_, timerId.seq_);
+  ActiveTimerSet::iterator it = activeTimers_.find(timer);
+  if (it != activeTimers_.end())
+  {
+    size_t n = timers_.erase(Entry(it->first->expiration(), it->first));
+    assert(n == 1); (void)n;
+    delete it->first; // FIXME: no delete please
+    activeTimers_.erase(it);
+  }
+  else if (callingExpiredTimers_)
+  {
+    cancelingTimers_.insert(timer);
+  }
+  assert(timers_.size() == activeTimers_.size());
+}
+
 void TimerQueue::handleRead()
 {
   loop_->assertInLoopThread();
@@ -138,25 +165,38 @@ void TimerQueue::handleRead()
 
   std::vector<Entry> expired = getExpired(now);
 
+  callingExpiredTimers_ = true;
+  cancelingTimers_.clear();
   // safe to callback outside critical section
   for (std::vector<Entry>::iterator it = expired.begin();
       it != expired.end(); ++it)
   {
     it->second->run();
   }
+  callingExpiredTimers_ = false;
 
   reset(expired, now);
 }
 
 std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now)
 {
+  assert(timers_.size() == activeTimers_.size());
   std::vector<Entry> expired;
-  Entry sentry = std::make_pair(now, reinterpret_cast<Timer*>(UINTPTR_MAX));
-  TimerList::iterator it = timers_.lower_bound(sentry);
-  assert(it == timers_.end() || now < it->first);
-  std::copy(timers_.begin(), it, back_inserter(expired));
-  timers_.erase(timers_.begin(), it);
+  Entry sentry(now, reinterpret_cast<Timer*>(UINTPTR_MAX));
+  TimerList::iterator end = timers_.lower_bound(sentry);
+  assert(end == timers_.end() || now < end->first);
+  std::copy(timers_.begin(), end, back_inserter(expired));
+  timers_.erase(timers_.begin(), end);
 
+  for (std::vector<Entry>::iterator it = expired.begin();
+      it != expired.end(); ++it)
+  {
+    ActiveTimer timer(it->second, it->second->sequence());
+    size_t n = activeTimers_.erase(timer);
+    assert(n == 1); (void)n;
+  }
+
+  assert(timers_.size() == activeTimers_.size());
   return expired;
 }
 
@@ -167,15 +207,17 @@ void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)
   for (std::vector<Entry>::const_iterator it = expired.begin();
       it != expired.end(); ++it)
   {
-    if (it->second->repeat())
+    ActiveTimer timer(it->second, it->second->sequence());
+    if (it->second->repeat()
+        && cancelingTimers_.find(timer) == cancelingTimers_.end())
     {
       it->second->restart(now);
       insert(it->second);
     }
     else
     {
       // FIXME move to a free list
-      delete it->second;
+      delete it->second; // FIXME: no delete please
     }
   }
 
@@ -192,15 +234,27 @@ void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)
 
 bool TimerQueue::insert(Timer* timer)
 {
+  loop_->assertInLoopThread();
+  assert(timers_.size() == activeTimers_.size());
   bool earliestChanged = false;
   Timestamp when = timer->expiration();
   TimerList::iterator it = timers_.begin();
   if (it == timers_.end() || when < it->first)
   {
     earliestChanged = true;
   }
-  std::pair<TimerList::iterator, bool> result = timers_.insert(std::make_pair(when, timer));
-  assert(result.second); (void)result;
+  {
+    std::pair<TimerList::iterator, bool> result
+      = timers_.insert(Entry(when, timer));
+    assert(result.second); (void)result;
+  }
+  {
+    std::pair<ActiveTimerSet::iterator, bool> result
+      = activeTimers_.insert(ActiveTimer(timer, timer->sequence()));
+    assert(result.second); (void)result;
+  }
+
+  assert(timers_.size() == activeTimers_.size());
   return earliestChanged;
 }
 

muduo/net/TimerQueue.h

@@ -49,15 +49,18 @@ class TimerQueue : boost::noncopyable
                    Timestamp when,
                    double interval);
 
-  // void cancel(TimerId timerId);
+  void cancel(TimerId timerId);
 
  private:
 
   // FIXME: use unique_ptr<Timer> instead of raw pointers.
   typedef std::pair<Timestamp, Timer*> Entry;
   typedef std::set<Entry> TimerList;
+  typedef std::pair<Timer*, int64_t> ActiveTimer;
+  typedef std::set<ActiveTimer> ActiveTimerSet;
 
   void scheduleInLoop(Timer* timer);
+  void cancelInLoop(TimerId timerId);
   // called when timerfd arms
   void handleRead();
   // move out all expired timers
@@ -71,6 +74,11 @@ class TimerQueue : boost::noncopyable
   Channel timerfdChannel_;
   // Timer list sorted by expiration
   TimerList timers_;
+
+  // for cancel()
+  ActiveTimerSet activeTimers_;
+  bool callingExpiredTimers_; /* atomic */
+  ActiveTimerSet cancelingTimers_;
 };
 
 }

muduo/net/tests/TimerQueue_unittest.cc

@@ -28,6 +28,12 @@ void print(const char* msg)
   }
 }
 
+void cancel(TimerId timer)
+{
+  g_loop->cancel(timer);
+  printf("cancelled at %s\n", Timestamp::now().toString().c_str());
+}
+
 int main()
 {
   printTid();
@@ -41,8 +47,12 @@ int main()
     loop.runAfter(1.5, boost::bind(print, "once1.5"));
     loop.runAfter(2.5, boost::bind(print, "once2.5"));
     loop.runAfter(3.5, boost::bind(print, "once3.5"));
+    TimerId t45 = loop.runAfter(4.5, boost::bind(print, "once4.5"));
+    loop.runAfter(4.2, boost::bind(cancel, t45));
+    loop.runAfter(4.8, boost::bind(cancel, t45));
     loop.runEvery(2, boost::bind(print, "every2"));
-    loop.runEvery(3, boost::bind(print, "every3"));
+    TimerId t3 = loop.runEvery(3, boost::bind(print, "every3"));
+    loop.runAfter(9.001, boost::bind(cancel, t3));
 
     loop.loop();
     print("main loop exits");