注：本文为阅读 muduo 源码库和作者著作之后的网络库复现和笔记

在传统的 IO 多路复用系统中，定时操作通常是直接去设置poll()等函数的超时时间，系统超时之后去执行对应的定时回调。但现代 Linux 系统已经将时间也抽象为了一种事件，并提供了对应的文件描述符fd机制，其能够被poll()等多路复用函数进行监视，因此可以很简洁地融入到我们之前实现的 Reactor 机制中。这也是更科学的做法。

底层 API

  • int timerfd_create(int clockid, int flags)
    根据用户指定的标识符生成 timer，并返回对应的fd。clockid设置时钟的类型，flags则是设置fd的各个标志位，比如是否阻塞等等。

  • int timerfd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value)
    通过new_value去设定 timer 的超时时间（expiration time）以及是否重复（repeat via same iterval）。需要注意的是，new_value默认情况下指的是基于 timer 当前时刻的相对时间，如果需要设置绝对时间，则需要更改flags参数。

模块设计

定时设置，也就是用户指定一个时刻和一个回调函数，系统需要在该时刻去执行该回调函数。这个定义包含了两层含义：

  • 用户的定时设置是任意的。有可能先设置一个 10 秒的定时，接下来再设置一个 5 秒的定时。用户不一定会以 expiration time 的先后顺序去设置定时；
  • 系统顺序执行回调函数。系统需要将所有注册过的定时事件，按照 expiration time 的先后顺序，在用户设置的每一个 expiration time 去执行对应的回调函数；

基于上述的特点，我们需要的定时器模块的流程示意如图 1 所示。
图1. 定时器任务流程

首先解释一下图 1 所示的流程：

  • class Timer: 显然每一个 expiration time 都对应着一个 callback function，因此将其封装为一个类class Timer，方便执行各自的回调函数。在图中即对应着五个不同颜色的小圆圈；
  • 接口：TimerQueue 向用户提供Timer Insert接口，供用户插入各个 Timer。在图中，我们先后插入了 expiration time 为 10s, 18s, 20s, 15s, 10.01s 的 Timer；
  • class TimerQueue内部过程：① 将得到的所有 Timer 进行排序；② 每当有新的 Timer expired 之时，抽取出过期的 Timer，在图中对应着 10s 和 10.01s 两个 Timer；③ 依次执行这些 expired Timer 的回调函数；

接下来则是根据流程所示的特点去分析如何构建class TimerQueue:

首先是何时执行timerfd_create()。这个很简单，自然是在构造函数中执行，将得到的 timer fd 作为class TimerQueue的成员变量；

其次是何时执行timerfd_settime()。这个是需要重点思考的问题。最直接的解决方案自然是每插入一个 Timer 就执行一次timerfd_settime()，对应图 1 的例子，系统就会将 10s, 10.01s, 15s, 18s, 20s 这几个时刻都设置为过期时刻，随后在每一个过期时刻将对应的 Timer 取出，并执行回调函数。但这个方案有一个潜在的问题。timerfd_settime()是系统调用，频繁的调用会导致系统开销增大。
改进的思路就是只对最早过期的 Timer 设置timerfd_settime()。具体而言分两种情况：

  • 新插入一个 Timer：在插入 Timer 之前比较该 Timer 和 TimerQueue 中排名最靠前的 Timer 的 expiration time，如果在其之前，则增设一个timerfd_settime()；
  • 取出过期的 Timer：在取出了过期的 Timer 之后，如果 TimerQueue 中还有剩余的 Timer，则再对残余的 Timer 中最靠前的那一个 Timer 设置timerfd_settime()

随后是如何组织 TimerQueue 中的 Timer。我们必须保证有序，而插入新 Timer 的位置又是随机的。综合考虑，使用std::set是最方便的。问题是如果存在 expiration time 相同的 Timer 怎么办。muduo 源码给出的解决方法很巧妙，其将std::set的元素设置为std::pair<TimeStamp, Timer *>，这样一来就完美的解决了问题，即使两个 Timer 的过期时刻一致，在std::set中也还能用 Timer 的地址去排列各个 Timer 的先后顺序。

最后就是将class TimerQueue集成到 EventLoop 中，也就是作为class EventLoop的成员变量，这个很简单，不再赘述。

实现过程中的一些知识点总结

  1. CLOCK_REALTIME和CLOCK_MONOTONIC
    在使用timer_create()中需要使用这些标志符去设定clockid(当然还有一些其他标识符可供选择)。这两个标识符的区别是：CLOCK_REALTIME基于实际的系统时间，可设定，非单调；CLOCK_MONOTONIC基于的是一个不确定的时间点，并单调流逝，其不可设定，绝对单调。为什么说CLOCK_REALTIME是非单调的呢？原因很简单，因为系统时间本身就是可以被用户任何修改的。具体详见博客1。

代码实战

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/* TimerQueue.h */
#ifndef TIMERQUEUE_H
#define TIMERQUEUE_H

#include <muduo/base/Timestamp.h>
#include <set>
#include <vector>
#include <memory>
#include <boost/noncopyable.hpp>

#include "Channel.h"

// forward declaration
class EventLoop;
class Timer;
class TimerId;

class TimerQueue : boost::noncopyable
{
  using Entry = std::pair<muduo::Timestamp, Timer *>;
  using TimerList = std::set<Entry>;
  using TimerCallback = std::function<void()>;
private:
  EventLoop *loop_;
  const int timerfd_;
  TimerList timers_;
  Channel timerfdChannel_;

  std::vector<Entry> getExpired(muduo::Timestamp now);
  bool insert(Timer *);
  void resetTimerfd(muduo::Timestamp when);


public:
  TimerQueue(EventLoop *);
  ~TimerQueue();

  TimerId addTimer(const TimerCallback cb, muduo::Timestamp time, double interval);
  void addTimerInLoop(Timer *timer);
  void handleRead();
  void reset(std::vector<Entry> &expired, muduo::Timestamp now);

};

#endif /* TIMERQUEUE_H */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
/* TimerQueue.cc */
#include <sys/timerfd.h>
#include <unistd.h>
#include "TimerQueue.h"
#include "Channel.h"
#include "Timer.h"
#include "EventLoop.h"
#include "TimerId.h"

int creatTimerFd(){
  return ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
}

struct timespec getTimeDiffFromNow(muduo::Timestamp when){
  auto microSeconds = when.microSecondsSinceEpoch()
                      - muduo::Timestamp::now().microSecondsSinceEpoch();
  if(microSeconds < 100)
    microSeconds = 100; // set the minimal diff as 100 micro seconds
  struct timespec ret;
  // seconds
  ret.tv_sec = static_cast<time_t>(microSeconds / muduo::Timestamp::kMicroSecondsPerSecond);
  // nano seconds
  ret.tv_nsec = static_cast<__SYSCALL_SLONG_TYPE>( (microSeconds%muduo::Timestamp::kMicroSecondsPerSecond) * 1000 );
  return ret;
}

std::vector<TimerQueue::Entry> TimerQueue::getExpired(muduo::Timestamp now){
  Entry dummy{now, reinterpret_cast<Timer *>(UINTPTR_MAX)};
  auto end_ = timers_.lower_bound(dummy);
  assert(end_==timers_.end() || now<end_->first);
  auto ret = std::vector<Entry>(timers_.begin(), end_);
  timers_.erase(timers_.begin(), end_);
  return ret;
}

TimerId TimerQueue::addTimer(const TimerCallback cb,
                          muduo::Timestamp time,
                          double interval){
  Timer *timer = new Timer(cb, time, interval);
  loop_->runInLoop(std::bind(&TimerQueue::addTimerInLoop, this, timer));
  return TimerId(timer, timer->sequence());
}

void TimerQueue::addTimerInLoop(Timer *timer){
  loop_->assertInLoopThread();
  auto isEarliest = insert(timer);
  if(isEarliest){
    resetTimerfd(timer->expiration());
    //printf("addtimer done\n");
  }
}

TimerQueue::TimerQueue(EventLoop *loop):
  loop_(loop),
  timerfd_(creatTimerFd()),
  timers_(),
  timerfdChannel_(loop, timerfd_)
{
  timerfdChannel_.setReadCallback(std::bind(&TimerQueue::handleRead, this));
  timerfdChannel_.enableRead(); // update timerfd Channel to Poller
  //printf("::TimerQueue done\n");
}

TimerQueue::~TimerQueue(){
  ::close(timerfd_);
  for(auto itr = timers_.begin(); itr != timers_.end(); ++itr){
    delete itr->second;
  }
}

void TimerQueue::handleRead(){
  loop_->assertInLoopThread();
  muduo::Timestamp now(muduo::Timestamp::now());
  auto expiredList = getExpired(now);
  for(auto &expired: expiredList){
    expired.second->run();
  }
  //printf("runned expired timer\n");
  reset(expiredList, now);
  //printf("reset done\n");
}

void TimerQueue::reset(std::vector<Entry> &expired, muduo::Timestamp now){
  muduo::Timestamp when;
  for(auto &e: expired){
    if(e.second->repeat()){
      e.second->restart(now); // addTimer(now, interval);
      insert(e.second);
    }
  }
  if(!timers_.empty()){
    when = timers_.begin()->first;
  } else {
    when = muduo::Timestamp::invalid();
  }
  if(when.valid()){
    resetTimerfd(when); // set the timer only when timers_ is not empty
    // and the next alarm time is the first element of timers_
  }
}

bool TimerQueue::insert(Timer * timer){
  bool isEarliset = false;
  auto itr = timers_.begin();
  if(itr==timers_.end() || timer->expiration()<itr->first){
    isEarliset = true;
  }
  timers_.insert({timer->expiration(), timer});
  return isEarliset;
}

void TimerQueue::resetTimerfd(muduo::Timestamp when){
  itimerspec old_timerspec, new_timerspec;
  bzero(&old_timerspec, sizeof old_timerspec);
  bzero(&new_timerspec, sizeof new_timerspec);
  new_timerspec.it_value = getTimeDiffFromNow(when);
  //printf("timer diff from now is %ld\n", new_timerspec.it_value.tv_sec);
  ::timerfd_settime(timerfd_, 0, &new_timerspec, &old_timerspec);;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
/* Timer.h */
#ifndef TIMER_H
#define TIMER_H

#include <functional>
#include <muduo/base/Timestamp.h>
#include <muduo/base/Atomic.h>
#include <boost/noncopyable>

class Timer : boost::noncopyable
{
  using TimerCallback = std::function<void()>;
private:
  TimerCallback callback_;
  muduo::Timestamp expiration_;
  double interval_;
  bool repeat_;
  int64_t sequence_;
  static muduo::AtomicInt64 sequenceNumGenerator_;

public:
  Timer(const TimerCallback &cb, muduo::Timestamp when, double itv);
  ~Timer();

  muduo::Timestamp expiration() const {
    return expiration_;
  }
  double interval() {
    return interval_;
  }
  int64_t sequence() const {
    return sequence_;
  }
  bool repeat() const {
    return repeat_;
  }

  void run() {
    //printf("run callback\n");
    callback_();
    //printf("callback runned\n");
  }

  void restart(muduo::Timestamp when);
};

#endif /* TIMER_H */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/* Timer.cc */
#include "Timer.h"

muduo::AtomicInt64 Timer::sequenceNumGenerator_;

Timer::Timer(const TimerCallback &cb,
             muduo::Timestamp when,
             double itv) :
  callback_(cb), expiration_(when), interval_(itv), repeat_(itv>0),
  sequence_(sequenceNumGenerator_.incrementAndGet())
{

}

Timer::~Timer(){

}

void Timer::restart(muduo::Timestamp when){
  if(repeat_){
    expiration_ = muduo::addTime(when, interval_);
  } else {
    expiration_ = muduo::Timestamp::invalid();
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
/* TimerId.h */
#ifndef TIMERID_H
#define TIMERID_H

#include <cstdint>

class Timer;

class TimerId
{
private:
  Timer *timer_;
  int64_t sequence_;

public:
  TimerId():
    timer_(nullptr),
    sequence_(0){

  }
  TimerId(Timer *t, int64_t sequ):
    timer_(t), sequence_(sequ) {

  }
};

#endif /* TIMERID_H */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/* main.cc */
#include "EventLoop.h"
#include <muduo/base/Thread.h>
#include "TimerId.h"

EventLoop *g_loop;

void timeOutFunc(){
  printf("Oooops, time out!!!\n");
  g_loop->quit();
}

void threadFunc(){
  g_loop->runAfter(15, &timeOutFunc);
}

int main(int argc, char *argv[])
{
  EventLoop loop;
  g_loop = &loop;
  muduo::Thread thread_(&threadFunc);
  thread_.start();
  loop.loop();
  return 0;
}

结果演示

在TimerQueue中注册一个Timer

Reference