1
2
3
4
5
6

Channel::Channel(EventLoop* loop, int fd__)
  : loop_(loop),
    fd_(fd__),
    ... {

}

1
2
3

const int Channel::kNoneEvent = 0;                //没有事件：0
const int Channel::kReadEvent = POLLIN | POLLPRI; //可读事件：3
const int Channel::kWriteEvent = POLLOUT;         //可写事件：4

1
2
3
4
5
6

/* Event types that can be polled for.  These bits may be set in `events'
   to indicate the interesting event types; they will appear in `revents'
   to indicate the status of the file descriptor.  */
#define POLLIN		0x001		/* There is data to read.  */
#define POLLPRI		0x002		/* There is urgent data to read.  */
#define POLLOUT		0x004		/* Writing now will not block.  */

1
2

int events_;  // 关注的事件
int revents_; // poll/epoll返回的事件，即实际监听到发生的事件类型

1
2
3
4

void enableReading() { events_ |= kReadEvent; update(); }
void enableWriting() { events_ |= kWriteEvent; update(); }
void disableWriting() { events_ &= ~kWriteEvent; update(); }
void disableAll() { events_ = kNoneEvent; update(); }

1
2
3
4
5

int fd() const { return fd_; }
int events() const { return events_; }
void set_revents(int revt) { revents_ = revt; } // used by pollers
// int revents() const { return revents_; }
bool isNoneEvent() const { return events_ == kNoneEvent; }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

typedef boost::function<void()> EventCallback;
typedef boost::function<void(Timestamp)> ReadEventCallback;

ReadEventCallback readCallback_;
EventCallback writeCallback_;
EventCallback closeCallback_;
EventCallback errorCallback_;

void setReadCallback(const ReadEventCallback &cb) { readCallback_ = cb; }
void setWriteCallback(const EventCallback &cb) { writeCallback_ = cb; }
void setCloseCallback(const EventCallback &cb) { closeCallback_ = cb; }
void setErrorCallback(const EventCallback &cb) { errorCallback_ = cb; }

 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

//处理绑定的事件
void Channel::handleEvent(Timestamp receiveTime) {
  boost::shared_ptr<void> guard;
  if (tied_) {
    guard = tie_.lock(); //确定shared_ptr指针还存在
    if (guard) {
      handleEventWithGuard(receiveTime);
    }
  } else {
    handleEventWithGuard(receiveTime);
  }
}

/**
 * poll函数的事件标志符值：
*    常量	                 说明
    POLLIN	        普通或优先级带数据可读
    POLLRDNORM    	普通数据可读
    POLLRDBAND    	优先级带数据可读
    POLLPRI	        高优先级数据可读
    POLLOUT	        普通数据可写
    POLLWRNORM	    普通数据可写
    POLLWRBAND	    优先级带数据可写
    POLLERR	        发生错误
    POLLHUP	        对方描述符挂起
    POLLNVAL	    描述字不是一个打开的文件
 *
 */
//开始选择性处理Channel对象绑定的事件
void Channel::handleEventWithGuard(Timestamp receiveTime) {
  eventHandling_ = true; //处于处理事件中

  // close
  if ((revents_ & POLLHUP) && !(revents_ & POLLIN)) {
    if (logHup_) {
      LOG_WARN << "Channel::handle_event() POLLHUP";
    }
    //符合关闭套接字的事件，回调close函数
    if (closeCallback_)
      closeCallback_();
  }

  //Invalid polling request.
  if (revents_ & POLLNVAL) {
    LOG_WARN << "Channel::handle_event() POLLNVAL";
  }

  // error
  if (revents_ & (POLLERR | POLLNVAL)) {
    if (errorCallback_)
      errorCallback_();
  }

  // read
  if (revents_ & (POLLIN | POLLPRI | POLLRDHUP)) {
    if (readCallback_)
      readCallback_(receiveTime);
  }

  // write
  if (revents_ & POLLOUT) {
    if (writeCallback_)
      writeCallback_();
  }

  eventHandling_ = false; //事件已处理完成
}

1
2
3

virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;
virtual void updateChannel(Channel* channel) = 0;
virtual void removeChannel(Channel* channel) = 0;

1
2
3
4
5
6
7

Poller *Poller::newDefaultPoller(EventLoop *loop) {
  if (::getenv("MUDUO_USE_POLL")) {
    return new PollPoller(loop);
  } else {
    return new EPollPoller(loop);
  }
}

1
2
3
4
5

typedef std::vector<struct pollfd> PollFdList;
typedef std::map<int, Channel*> ChannelMap;

PollFdList pollfds_;
ChannelMap channels_;

 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

/**
 * 主要时将Channel负责的文件描述符(fd)注册到Poller对象中, 或更新Poller对象
 */
void PollPoller::updateChannel(Channel *channel) {
  Poller::assertInLoopThread();
  LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events();

  if (channel->index() < 0) {
    // index < 0说明是一个新的通道
    // a new one, add to pollfds_
    assert(channels_.find(channel->fd()) == channels_.end());

    struct pollfd pfd;
    pfd.fd = channel->fd();
    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0; // channel->revents()
    pollfds_.push_back(pfd);

    int idx = static_cast<int>(pollfds_.size()) - 1;
    channel->set_index(idx);
    channels_[pfd.fd] = channel; //通过文件描述符(fd) 映射 Channel
  } else {
    // update existing one
    //已存在的Channel
    assert(channels_.find(channel->fd()) != channels_.end());
    assert(channels_[channel->fd()] == channel);

    // index的取值必定在channels_.size()内
    int idx = channel->index();
    assert(0 <= idx && idx < static_cast<int>(pollfds_.size()));

    /**
     * 更新pollfds_容器的信息(struct pollfd)
     */
    struct pollfd &pfd = pollfds_[idx];

    // pfd.fd == -1 代表此Chanel不需被Poller::poll检测
    assert(pfd.fd == channel->fd() || pfd.fd == -channel->fd() - 1);

    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0;

    //如果不再关注该Channel中文件描述符事件，则直接将该文件描述符赋值为其相反数减一。
    // 将一个通道暂时更改为不关注事件，但不从Poller中移除该通道
    if (channel->isNoneEvent()) {
      // ignore this pollfd
      // 暂时忽略该文件描述符的事件
      // 这里pfd.fd 可以直接设置为-1
      pfd.fd = -channel->fd() - 1; // 这样子设置是为了removeChannel优化
    }
  }
}

 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

void PollPoller::removeChannel(Channel *channel) {
  Poller::assertInLoopThread();
  LOG_TRACE << "fd = " << channel->fd();

  assert(channels_.find(channel->fd()) != channels_.end());
  assert(channels_[channel->fd()] == channel);
  assert(channel->isNoneEvent());

  int idx = channel->index();

  assert(0 <= idx && idx < static_cast<int>(pollfds_.size()));

  const struct pollfd &pfd = pollfds_[idx];
  (void)pfd;
  assert(pfd.fd == -channel->fd() - 1 && pfd.events == channel->events());

  size_t n = channels_.erase(channel->fd());
  assert(n == 1);
  (void)n;
  
  if (implicit_cast<size_t>(idx) == pollfds_.size() - 1) {
    pollfds_.pop_back();
  } else {
    // 这里移除的算法复杂度是O(1)，将待删除元素与最后一个元素交换再pop_back
    int channelAtEnd = pollfds_.back().fd;
    iter_swap(pollfds_.begin() + idx, pollfds_.end() - 1);

    //不再关注该Channel中文件描述符事件，通过这种方式获取原有的下标(key)位置
    if (channelAtEnd < 0) {
      channelAtEnd = -channelAtEnd - 1;
    }
    channels_[channelAtEnd]->set_index(idx);
    pollfds_.pop_back();
  }
}

1

typedef std::vector<Channel*> ChannelList;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

/**
 * poll机制与select机制类似，通过管理文件描述符来进行轮询，效率更高，并且处理的连接个数不受内核的限制。
 * 通过poll的方式进行轮询，通过Poller::fillActiveChannels得到活跃事件
 */
Timestamp PollPoller::poll(int timeoutMs, ChannelList *activeChannels) {
  // XXX pollfds_ shouldn't change
  int numEvents = ::poll(&*pollfds_.begin(), pollfds_.size(), timeoutMs);

  Timestamp now(Timestamp::now());

  if (numEvents > 0) {
    LOG_TRACE << numEvents << " events happended";
    fillActiveChannels(numEvents, activeChannels);
  } else if (numEvents == 0) {
    LOG_TRACE << " nothing happended";
  } else {
    LOG_SYSERR << "PollPoller::poll()";
  }

  return now;
}

 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

/**
 * struct pollfd {
    int fd;         	  // 用于检测的文件描述符
    short events;         // 等待的事件类型
    short revents;        // 实际发生的事件类型
  } ;

  通过fd 获得发生事件的Chanel集合
**/
void PollPoller::fillActiveChannels(int numEvents,
                                    ChannelList *activeChannels) const {
  for (PollFdList::const_iterator pfd = pollfds_.begin();
       pfd != pollfds_.end() && numEvents > 0; ++pfd) {
    if (pfd->revents > 0) {
      --numEvents;
      ChannelMap::const_iterator ch = channels_.find(pfd->fd);
      assert(ch != channels_.end());

      Channel *channel = ch->second;
      assert(channel->fd() == pfd->fd);

      channel->set_revents(pfd->revents);
      // pfd->revents = 0;
      activeChannels->push_back(channel);
    }
  }
}

1
2
3
4
5

typedef std::vector<struct epoll_event> EventList;
typedef std::map<int, Channel*> ChannelMap;

EventList events_;
ChannelMap channels_;

 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

//将Channel加入epoll中管理，updateChannel方法一定是I/O线程调用的
void EPollPoller::updateChannel(Channel *channel) {
  Poller::assertInLoopThread();
  LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events();

  const int index = channel->index();
  if (index == kNew || index == kDeleted) {
    // a new one, add with EPOLL_CTL_ADD
    int fd = channel->fd();

    if (index == kNew) {
      assert(channels_.find(fd) == channels_.end());
      channels_[fd] = channel;
    } else // index == kDeleted
    {
      //也就是说，当Channel对象的无事件，可以通过index_来标识为删除状态，并不会从channels_容器中移除
      assert(channels_.find(fd) != channels_.end());
      assert(channels_[fd] == channel);
    }
    channel->set_index(kAdded);     //设置Channel对象为添加状态
    update(EPOLL_CTL_ADD, channel); //添加到epoll中监视
  } else {
    // update existing one with EPOLL_CTL_MOD/DEL
    //更新已存在的Channel描述符，修改或删除
    int fd = channel->fd();
    (void)fd;

    assert(channels_.find(fd) != channels_.end());
    assert(channels_[fd] == channel);
    assert(index == kAdded);

    //当Channel对象不绑定事件，那么就从epoll的监视中移除，并设置Channel对象为删除状态
    if (channel->isNoneEvent()) {
      update(EPOLL_CTL_DEL, channel);
      channel->set_index(kDeleted);
    } else { //更新epoll中监视的Channel对象的信息
      update(EPOLL_CTL_MOD, channel);
    }
  }
}

1
2
3
4
5

namespace {
const int kNew = -1;        //还没有和EPollPoller关联
const int kAdded = 1;       //该Channel已经和EPollPoller关联
const int kDeleted = 2;     //该通道被取消关注
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

//更新Channel对象绑定的事件
void EPollPoller::update(int operation, Channel *channel) {
  struct epoll_event event;
  bzero(&event, sizeof event);
  event.events = channel->events();
  event.data.ptr = channel;

  int fd = channel->fd();
  if (::epoll_ctl(epollfd_, operation, fd, &event) < 0) {
    if (operation == EPOLL_CTL_DEL) {
      LOG_SYSERR << "epoll_ctl op=" << operation << " fd=" << fd;
    } else {
      LOG_SYSFATAL << "epoll_ctl op=" << operation << " fd=" << fd;
    }
  }
}

 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

// 将Channel对象从channels_容器和epoll监控中移除，此操作必须是I/O线程调用的
void EPollPoller::removeChannel(Channel *channel) {
  Poller::assertInLoopThread();

  int fd = channel->fd();
  LOG_TRACE << "fd = " << fd;

  assert(channels_.find(fd) != channels_.end());
  assert(channels_[fd] == channel);
  assert(channel->isNoneEvent()); //要移除的Channel对象本身一定是无事件的

  int index = channel->index();

  // 要移除的Channel对象应该就可能处于关联状态、也可能处于被取消关注状态(kDeleted)
  assert(index == kAdded || index == kDeleted);

  size_t n = channels_.erase(fd);
  (void)n;
  assert(n == 1);

  // 处于关联状态，说明epoll还在监视它
  if (index == kAdded) {
    update(EPOLL_CTL_DEL, channel);
  }

  // 重新设置Channel为未被管理的初始状态
  channel->set_index(kNew);
}

1

typedef std::vector<Channel*> ChannelList;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

Timestamp EPollPoller::poll(int timeoutMs, ChannelList *activeChannels) {
  //活跃的事件数量
  int numEvents = ::epoll_wait(epollfd_, &*events_.begin(),
                               static_cast<int>(events_.size()), timeoutMs);

  Timestamp now(Timestamp::now());

  if (numEvents > 0) {
    LOG_TRACE << numEvents << " events happended";
    fillActiveChannels(numEvents, activeChannels);

    //活跃的事件数量太多了，扩容，动态调整
    if (implicit_cast<size_t>(numEvents) == events_.size()) {
      events_.resize(events_.size() * 2);
    }
  } else if (numEvents == 0) {
    LOG_TRACE << " nothing happended";
  } else {
    // numEvents < 0 的情况
    LOG_SYSERR << "EPollPoller::poll()";
  }
  return now;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

void EPollPoller::fillActiveChannels(int numEvents,
                                     ChannelList *activeChannels) const {
  //活跃的事件数量不可能比装所有事件的容器的元素数量多
  assert(implicit_cast<size_t>(numEvents) <= events_.size());

  for (int i = 0; i < numEvents; ++i) {
    Channel *channel = static_cast<Channel *>(events_[i].data.ptr);

// NDEBUG模式
#ifndef NDEBUG
    int fd = channel->fd();
    ChannelMap::const_iterator it = channels_.find(fd);
    assert(it != channels_.end());
    assert(it->second == channel);
#endif

    channel->set_revents(events_[i].events);
    activeChannels->push_back(channel);
  }
}

1

EPollPoller::~EPollPoller() { ::close(epollfd_); }

1
2
3
4
5
6
7

#include <sys/timerfd.h>

int timerfd_create(int clockid, int flags);

int timerfd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value);

int timerfd_gettime(int fd, struct itimerspec *curr_value)

1
2
3
4
5
6
7
8
9

struct timespec {
    time_t tv_sec;      /* Seconds */
    long   tv_nsec;     /* Nanoseconds */
};

struct itimerspec {
    struct timespec it_interval;  /* Interval for periodic timer */
    struct timespec it_value;     /* Initial expiration */
};

1
2
3
4
5

const TimerCallback callback_;  //定时执行函数
Timestamp expiration_;          // 超时时间
const double interval_;         // 执行间隔
const bool repeat_;             // 是否重复定时执行任务
const int64_t sequence_;        // 序列号，每个Tmier的序列号都不一样

1

static AtomicInt64 s_numCreated_;

1
2
3
4
5
6
7
8

Timer(const TimerCallback& cb, Timestamp when, double interval)
  : callback_(cb),
    expiration_(when),
    interval_(interval),
    repeat_(interval > 0.0),
    sequence_(s_numCreated_.incrementAndGet()) {

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

  void run() const { callback_(); }

  //触发事件的时间戳
  Timestamp expiration() const { return expiration_; }

  bool repeat() const { return repeat_; }

  int64_t sequence() const { return sequence_; }

  void restart(Timestamp now);

  static int64_t numCreated() { return s_numCreated_.get(); }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

/**
 * 如果是重复的累加的，加入累加的时间重新设置到期时间，否则设置为失效时间
 */
void Timer::restart(Timestamp now) {
  if (repeat_) {
    // 重新计算下一个超时时刻
    expiration_ = addTime(now, interval_);
  } else {
    //设置无效的时间
    expiration_ = Timestamp::invalid();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class Timer;

///
/// An opaque identifier, for canceling Timer.
///
class TimerId : public muduo::copyable {
public:
  TimerId() : timer_(NULL), sequence_(0) {}

  TimerId(Timer *timer, int64_t seq) : timer_(timer), sequence_(seq) {}

  // default copy-ctor, dtor and assignment are okay

  friend class TimerQueue;

private:
  Timer *timer_;
  int64_t sequence_;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

// FIXME: use unique_ptr<Timer> instead of raw pointers.
// unique_ptr是C++ 11标准的一个独享所有权的智能指针
// 无法得到指向同一对象的两个unique_ptr指针
// 但可以进行移动构造与移动赋值操作，即所有权可以移动到另一个对象(而非拷贝构造)
typedef std::pair<Timestamp, Timer *> Entry;
typedef std::set<Entry> TimerList;
typedef std::pair<Timer *, int64_t> ActiveTimer;
typedef std::set<ActiveTimer> ActiveTimerSet;

EventLoop *loop_; // 所属EventLoop
const int timerfd_;
Channel timerfdChannel_;
// Timer list sorted by expiration
// set中的元素都是排好序的, 用于存储Timer，同时将Timer的到期时间戳作为key
TimerList timers_; // timers_是按到期时间排序

// for cancel()
// timers_与activeTimers_保存的是相同的数据 =》 利于操作，空间换效率吧？？？
// timers_是按到期时间排序，activeTimers_是按对象地址排序
ActiveTimerSet activeTimers_;
bool callingExpiredTimers_;      //用来确定是否正在调用回调函数的定时器/* atomic */
ActiveTimerSet cancelingTimers_; // 保存的是被取消的定时器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

/**
 * 一个TimerQueue只对应一个Eventloop和一个内部构造的Channel(构造对象时，默认绑定一个定时器对象timerfd_)
 * 同时设置Channel的回调函数TimerQueue::handleRead
 */
TimerQueue::TimerQueue(EventLoop *loop)
    : loop_(loop), timerfd_(createTimerfd()), timerfdChannel_(loop, timerfd_),
      timers_(), callingExpiredTimers_(false) {
  timerfdChannel_.setReadCallback(boost::bind(&TimerQueue::handleRead, this));
  // we are always reading the timerfd, we disarm it with timerfd_settime.
  timerfdChannel_.enableReading();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

/**
 * 创建定时器，生成一个定时器对象，返回与之关联的文件描述符timerfd。
 */
int createTimerfd() {
  int timerfd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
  if (timerfd < 0) {
    LOG_SYSFATAL << "Failed in timerfd_create";
  }
  return timerfd;
}

 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

/**
 * I/O线程才可调用
 * 用来设置关联的Channel的回调函数
 */
void TimerQueue::handleRead() {
  loop_->assertInLoopThread();

  Timestamp now(Timestamp::now());
  // 清除该事件，避免一直触发，并记录触发TimerQueue::handleRead回调函数的时间
  readTimerfd(timerfd_, now);

  // 获取该时刻之前所有的定时器列表(即超时定时器列表)
  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);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// 清除定时器，避免一直触发，并记录传入的时间戳
void readTimerfd(int timerfd, Timestamp now) {
  uint64_t howmany;
  ssize_t n = ::read(timerfd, &howmany, sizeof howmany);
  LOG_TRACE << "TimerQueue::handleRead() " << howmany << " at "
            << now.toString();

  if (n != sizeof howmany) {
    LOG_ERROR << "TimerQueue::handleRead() reads " << n
              << " bytes instead of 8";
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

/**
 * 添加一个新Timer，并及时更新；
 * 如果TimerList中元素只有一个或插入的Timer最快到期，马上重新设置TimerQueue对象绑定的timerfd的超时时间
 */
TimerId TimerQueue::addTimer(const TimerCallback &cb, Timestamp when,
                             double interval) {
  /**
   * 该对象将存储到TimerList中
   * => std::set<std::pair<Timestamp, Timer*>> TimerList;
   */
  Timer *timer = new Timer(cb, when, interval);

  loop_->runInLoop(boost::bind(&TimerQueue::addTimerInLoop, this, timer));

  // addTimerInLoop(timer);
  return TimerId(timer, timer->sequence());
}

1

typedef boost::function<void()> TimerCallback;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

/**
 * I/O线程才可调用
 */
void TimerQueue::addTimerInLoop(Timer *timer) {
  loop_->assertInLoopThread();
  // 插入一个定时器，有可能会使得最早到期的定时器发生改变
  bool earliestChanged = insert(timer);

  if (earliestChanged) {
    // 重置定时器的超时时刻(timerfd_settime)
    resetTimerfd(timerfd_, timer->expiration());
  }
}

 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

/**
 * 返回值，插入的Timer是否是最早到期的时间;说明：插入的Timer元素是有序timers_(set集合)的第一个元素或唯一的元素为true
 * 功能：插入一个std::pair<Timestamp，Timer*>对象，将参数timer的到期时间戳作为std::pair的first，Timner对象作为sencond
 * activeTimers_ 和 timers_两个容器保存的Timer对象是一样的，仅仅内存分布不同而已
 */
bool TimerQueue::insert(Timer *timer) {
  loop_->assertInLoopThread();
  assert(timers_.size() == activeTimers_.size());

  // 最早到期时间是否改变
  bool earliestChanged = false;
  Timestamp when = timer->expiration();
  TimerList::iterator it = timers_.begin();

  // 如果timers_为空或者when小于timers_中的最早到期时间(这个插入的Timer是集合第一个到期的时间戳)
  if (it == timers_.end() || when < it->first) {
    earliestChanged = true;
  }

  {
    // 插入到timers_中
    std::pair<TimerList::iterator, bool> result =
        timers_.insert(Entry(when, timer));

    // result.first代表插入的元素的iterator，result.second代表元素是否插入成功
    assert(result.second);
    (void)result;
  }

  {
    // 插入到activeTimers_中
    std::pair<ActiveTimerSet::iterator, bool> result =
        activeTimers_.insert(ActiveTimer(timer, timer->sequence()));
    assert(result.second);
    (void)result;
  }

  assert(timers_.size() == activeTimers_.size());
  return earliestChanged;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

/**
 * 重置定时器的超时时间，重新设置Timer文件描述符(Timerfd)的超时时间，启动或停止定时器
 */
void resetTimerfd(int timerfd, Timestamp expiration) {
  // wake up loop by timerfd_settime()
  struct itimerspec newValue;
  struct itimerspec oldValue;
  bzero(&newValue, sizeof newValue);
  bzero(&oldValue, sizeof oldValue);
  newValue.it_value = howMuchTimeFromNow(expiration);

  //启动或停止定时器，依据newValue
  int ret = ::timerfd_settime(timerfd, 0, &newValue, &oldValue);
  if (ret) {
    LOG_SYSERR << "timerfd_settime()";
  }
}

 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

/**
 * rvo优化
 * 返回一个到期时间std::vector<Entry>的容器
 * =》typedef std::pair<Timestamp,Timer*> Entry;
 */
std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now) {
  assert(timers_.size() == activeTimers_.size());

  std::vector<Entry> expired;
  // std::pair作为set的key时，比较pair 是先比较first, first相同再比较second
  Entry sentry(now, reinterpret_cast<Timer *>(UINTPTR_MAX));

  // 返回第一个未到期的Timer的迭代器
  // lower_bound的含义是返回第一个值>=sentry的元素的iterator
  // 即*end >= sentry，从而end->first > now
  TimerList::iterator end = timers_.lower_bound(sentry);

  // timers_的集合全是过期的时间戳，或者？？？
  assert(end == timers_.end() || now < end->first);

  // 将到期的定时器插入到expired中
  std::copy(timers_.begin(), end, back_inserter(expired));
  // 从timers_中移除到期的定时器
  timers_.erase(timers_.begin(), end);

  // 从activeTimers_中移除到期的定时器
  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;
}

1
2
3

void TimerQueue::cancel(TimerId timerId) {
  loop_->runInLoop(boost::bind(&TimerQueue::cancelInLoop, this, timerId));
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

/**
 * I/O线程才可调用
 */
void TimerQueue::cancelInLoop(TimerId timerId) {
  loop_->assertInLoopThread();
  assert(timers_.size() == activeTimers_.size());

  ActiveTimer timer(timerId.timer_, timerId.sequence_);
  // 查找该定时器
  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,如果用了unique_ptr,这里就不需要手动删除了
    activeTimers_.erase(it);
  } else if (callingExpiredTimers_) {
    // 已经到期，并且正在调用回调函数的定时器
    cancelingTimers_.insert(timer);
  }
  assert(timers_.size() == activeTimers_.size());
}

1

ActiveTimerSet cancelingTimers_;

 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

/**
 * I/O线程才可调用
 * 用来设置关联的Channel的回调函数
 */
void TimerQueue::handleRead() {
  loop_->assertInLoopThread();

  Timestamp now(Timestamp::now());
  // 清除该事件，避免一直触发，并记录触发TimerQueue::handleRead回调函数的时间
  readTimerfd(timerfd_, now);

  // 获取该时刻之前所有的定时器列表(即超时定时器列表)
  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);
}

 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

/**
 * 对已经过期的队列进行处理，如果是重复调用的，重新设置新的到期时间并加入Timer队列中继续监控；重新设置TimerQueue对象绑定的fd到期时间
 */
void TimerQueue::reset(const std::vector<Entry> &expired, Timestamp now) {
  Timestamp nextExpire;

  for (std::vector<Entry>::const_iterator it = expired.begin();
       it != expired.end(); ++it) {
    ActiveTimer timer(it->second, it->second->sequence());
    // 如果是重复的定时器并且是未取消定时器，则重启该定时器
    if (it->second->repeat() &&
        cancelingTimers_.find(timer) == cancelingTimers_.end()) {
      //如果是重复的累加的，加入累加的时间重新设置到期时间，否则设置为失效时间
      it->second->restart(now);
      //将累加后新的Timer插入到Timer队列中，继续监控
      insert(it->second);
    } else {
      // 一次性定时器或者已被取消的定时器是不能重置的，因此删除该定时器
      // FIXME move to a free list
      delete it->second; // FIXME: no delete please
    }
  }

  if (!timers_.empty()) {
    // //得到队列中第一个要到期的时间，获取最早到期的定时器超时时间
    nextExpire = timers_.begin()->second->expiration();
  }

  //队列中还未到期的时间是有效的
  if (nextExpire.valid()) {
    // 重置定时器的超时时刻(timerfd_settime)，重新设置TimerQueue对象绑定的fd到期时间
    resetTimerfd(timerfd_, nextExpire);
  }
}

1
2
3
4
5
6
7

TimerQueue::~TimerQueue() {
  ::close(timerfd_); //关闭关联的描述符
  // do not remove channel, since we're in EventLoop::dtor();
  for (TimerList::iterator it = timers_.begin(); it != timers_.end(); ++it) {
    delete it->second;
  }
}

1
2
3
4
5
6
7
8

//将Chanel注册到Poller对象的polldfs数组中
void Channel::update() { loop_->updateChannel(this); }

void EventLoop::updateChannel(Channel *channel) {
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  poller_->updateChannel(channel);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 调用这个函数之前确保调用disableAll
void Channel::remove() {
  assert(isNoneEvent());
  loop_->removeChannel(this);
}

void EventLoop::removeChannel(Channel *channel) {
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  if (eventHandling_) {
    assert(currentActiveChannel_ == channel ||
           std::find(activeChannels_.begin(), activeChannels_.end(), channel) ==
               activeChannels_.end());
  }
  poller_->removeChannel(channel);
}

1
2
3
4
5
6
7
8
9

boost::scoped_ptr<TimerQueue> timerQueue_;

EventLoop::EventLoop()
  : looping_(false),
    ...
    timerQueue_(new TimerQueue(this)),
    ... {
    ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

///
/// Runs callback at 'time'.
/// Safe to call from other threads.
///
//直接回调
TimerId runAt(const Timestamp &time, const TimerCallback &cb);
///
/// Runs callback after @c delay seconds.
/// Safe to call from other threads.
///
//延期回调
TimerId runAfter(double delay, const TimerCallback &cb);
///
/// Runs callback every @c interval seconds.
/// Safe to call from other threads.
///
//以interval的频率回调
TimerId runEvery(double interval, const TimerCallback &cb);
///
/// Cancels the timer.
/// Safe to call from other threads.
///
void cancel(TimerId timerId);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

TimerId EventLoop::runAt(const Timestamp &time, const TimerCallback &cb) {
  return timerQueue_->addTimer(cb, time, 0.0);
}

TimerId EventLoop::runAfter(double delay, const TimerCallback &cb) {
  Timestamp time(addTime(Timestamp::now(), delay));
  return runAt(time, cb);
}

TimerId EventLoop::runEvery(double interval, const TimerCallback &cb) {
  Timestamp time(addTime(Timestamp::now(), interval));
  return timerQueue_->addTimer(cb, time, interval);
}

void EventLoop::cancel(TimerId timerId) { return timerQueue_->cancel(timerId); }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

/**
 * 添加一个新Timer，并及时更新；
 * 如果TimerList中元素只有一个或插入的Timer最快到期，马上重新设置TimerQueue对象绑定的timerfd的超时时间
 */
TimerId TimerQueue::addTimer(const TimerCallback &cb, Timestamp when,
                             double interval) {
  /**
   * 该对象将存储到TimerList中
   * std::set<std::pair<Timestamp, std::unique_ptr<Timer>>>
   */
  Timer *timer = new Timer(cb, when, interval);

  loop_->runInLoop(boost::bind(&TimerQueue::addTimerInLoop, this, timer));

  // addTimerInLoop(timer);
  return TimerId(timer, timer->sequence());
}

void TimerQueue::cancel(TimerId timerId) {
  loop_->runInLoop(boost::bind(&TimerQueue::cancelInLoop, this, timerId));

  // cancelInLoop(timerId);
}

1
2
3
4
5
6
7
8
9

boost::scoped_ptr<Poller> poller_;

EventLoop::EventLoop()
  : looping_(false),
    ...
    poller_(Poller::newDefaultPoller(this)),
    ... {
    ...
}

1
2
3
4
5
6
7

Poller *Poller::newDefaultPoller(EventLoop *loop) {
  if (::getenv("MUDUO_USE_POLL")) {
    return new PollPoller(loop);
  } else {
    return new EPollPoller(loop);
  }
}

 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

// 事件循环，该函数不能跨线程调用
// 只能在创建该对象的线程中调用
void EventLoop::loop() {
  assert(!looping_);
  // 断言当前处于创建该对象的线程中
  assertInLoopThread();
  looping_ = true;
  quit_ = false;
  LOG_TRACE << "EventLoop " << this << " start looping";

  //::poll(NULL, 0, 5*1000);
  while (!quit_) {
    activeChannels_.clear();
    //执行完Poller::poll()的时间
    pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
    //++iteration_;
    if (Logger::logLevel() <= Logger::TRACE) {
      printActiveChannels();
    }
    // TODO sort channel by priority
    eventHandling_ = true; //标识事件正在处理
    for (ChannelList::iterator it = activeChannels_.begin();
         it != activeChannels_.end(); ++it) {
      currentActiveChannel_ = *it;
      currentActiveChannel_->handleEvent(pollReturnTime_); //事件处理
    }
    currentActiveChannel_ = NULL;
    eventHandling_ = false; //标识事件已处理完成

    // 让IO线程也能执行一些计算任务，IO不忙的时候，处于阻塞状态
    doPendingFunctors(); // 执行其他线程或者本线程添加的一些回调任务
  }

  LOG_TRACE << "EventLoop " << this << " stop looping";
  looping_ = false;
}

1
2
3
4
5
6

bool callingPendingFunctors_; //判断是否IO线程处理一些回调任务中=>pendingFunctors_
int wakeupFd_; // 用于eventfd
// unlike in TimerQueue, which is an internal class,
// we don't expose Channel to client.
boost::scoped_ptr<Channel> wakeupChannel_; // 该通道将会纳入poller_来管理
std::vector<Functor> pendingFunctors_; // @BuardedBy mutex_   即将发生的回调，即在IO线程中执行需要执行回调函数集合