muduo源码分析之EventLoop、Channel、Poller的实现

作者一直强调的一个概念叫做one loop per thread，撇开多线程不谈，本篇博文将学习，怎么将传统的I/O复用poll/epoll封装到C++ 类中。

1.I/O复用

复习使用poll/epoll进行I/O复用的一些编程内容。

使用poll

对于一个文件描述符fd来说，我们将通过struct pollfd来设置我们关注的事件event，并在通过poll调用返回获取活跃的事件revent。比如说(伪代码)：

struct pollfd pfds[2];
fd1 = GET_FD();
fd2 = GET_FD();

pfds[0].fd=fd1;
pfds[0].events=POLLIN;
//fd2 is the same
while(true)
{
    ret=poll(pfds,pfds中关心的描述符数目,TIMEOUT);
    if(pfds[0].revents & POLLOUT)
    {
        ///
    }
    ///
}

使用epoll

epoll就要复杂一些了，主要由epoll_create、epoll_ctl、epoll_wait函数组成，比如说(伪代码)：

epollfd=epoll_create()
struct epoll_event ev;
ev.event=EPOLL_IN;
ev.data.fd=fd;
epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);
while(true)
{
    epoll_wait();
    handle_event();
}

2.什么都不做的EventLoop

本文代码参见https://github.com/chenshuo/recipes

one loop per thread表示每个线程只能有一个EventLoop对象，怎么来保证呢？
作者用__thread关键字修饰的t_loopInThisThread来存储对象指针（它的值在多线程环境下互不干扰），每当EventLoop进行构造的时候，判断该指针是否为NULL也就是在之前有没有被初始化过。

// reactor/s00/EventLoop.cc

/*__thread关键字 各个线程之间的值相互不干扰*/
__thread EventLoop* t_loopInThisThread = 0;

EventLoop::EventLoop()
  : looping_(false),
    threadId_(CurrentThread::tid())
{
  LOG_TRACE <<"EventLoop created " <<this <<" in thread " <  /*t_loopInThisThread不为空表示当前线程创建了EventLoop对象*/
  /*one thread per loop表示一个线程只能有一个EventLoop对象*/
  if (t_loopInThisThread)
  {

    LOG_FATAL <<"Another EventLoop " <              <<" exists in this thread " <  }
  else
  {
      /*记录该线程的EventLoop对象*/
    t_loopInThisThread = this;
  }
}

对于一个简单的EventLoop来说，poll调用封装在EventLoop::loop()中，比如说：

void EventLoop::loop()
{
  assert(!looping_);
  /*确保EventLoop在创建它的线程中loop调用*/
  assertInLoopThread();
  /*等待5秒的poll调用*/
  ::poll(NULL, 0, 5*1000);
  LOG_TRACE <<"EventLoop " <<this <<" stop looping";

}

3.Reactor关键结构

主要介绍Channel类（用于对描述符及事件的封装和事件的分发）Poller类（主要对poll / epoll调用的封装，但只是获取活跃事件并不负责事件分发）

我一直思考的问题就在于Channel、Poller、EventLoop是如何协调工作的，所以，先从一个示例看起。

// reactor/s01/test3.cc

muduo::EventLoop* g_loop;

void timeout()
{
  printf("Timeout!\n");
  g_loop->quit();
}

int main()
{
  muduo::EventLoop loop;//EventLoop对象
  g_loop = &loop;


//关心的描述符
  int timerfd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
  //创建channel对象，保存一份EventLoop的指针
  muduo::Channel channel(&loop, timerfd);
  //设置描述符可读时执行的回调函数
  channel.setReadCallback(timeout);
  //该函数用于设置关心的事件event
  //并且会依次调用Channel::update ---> EventLoop::updateChannel --->Poller::updateChannel
  channel.enableReading();

  struct itimerspec howlong;
  bzero(&howlong, sizeof howlong);
  howlong.it_value.tv_sec = 5;
  ::timerfd_settime(timerfd, 0, &howlong, NULL);

//loop循环
  loop.loop();

  ::close(timerfd);
}

程序执行的结果是5秒之后timefd可读将执行回调函数，并关闭loop。

现在我们来阅读相关源码，从上面的注释，首先，需要构造一个Channel对象：

Channel::Channel(EventLoop* loop, int fdArg)
  : loop_(loop),//保存EventLoop对象的指针
    fd_(fdArg),//描述符
    events_(0),//关心的事件
    revents_(0),//活跃的事件
    index_(-1)//index_用于记录当前fd在pollfd数组中的位置
{
}

之后设置回调，这里用到boost::function和boost::bind就不再详细描述。

enableReading干了两件事情：
1、设置关心的事情为POLL_IN，事件可读。
2、将fd添加到pollfd数组中。
第2步将依次调用Channel::update ---> EventLoop::updateChannel --->Poller::updateChannel
分别来看看这些函数：

void Channel::update()
{
  loop_->updateChannel(this);//这里将channel对象的指针传到了EventLoop里面
  /*loop->updateChannel--->Poller::updateChannel*/
}
/////////////////////////////////////////////////////////////////////////////
void EventLoop::updateChannel(Channel* channel)
{
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  //poller_是EventLoop中Poller类型的成员通过EventLoop的构造函数初始化。
  poller_->updateChannel(channel);//将channel对象的指针传给了Poller对象
}
/////////////////////////////////////////////////////////////////////////
void Poller::updateChannel(Channel* channel)
{
    /*与其ownerLoop在同一个线程*/
  assertInLoopThread();
  LOG_TRACE <<"fd = " <fd() <<" events = " <events();
  if (channel->index() <0) {//添加时Channel::index为初始化值-1
      /*添加一个新的*/
      /*每次都在尾部添加*/
    // a new one, add to pollfds_
    assert(channels_.find(channel->fd()) == channels_.end());
    struct pollfd pfd;//pollfd结构体
    //进行设置
    pfd.fd = channel->fd();
    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0;
    //pollfds_是Poller类中的成员是一个pollfd类型的 vector，充当poll调用的第一个参数
    pollfds_.push_back(pfd);
    int idx = static_cast<int>(pollfds_.size())-1;//重新计算索引，方便下次更新时快速定位
    channel->set_index(idx);//类似于设置上下文
    channels_[pfd.fd] = channel;//channels_是fd和Channel的一个map
  } else {
    // update existing one
    // 更新fd，由于channel保存有index，因此访问的时间效率为O(1)
    // 可能修改event后进入该分分支。
    assert(channels_.find(channel->fd()) != channels_.end());
    assert(channels_[channel->fd()] == channel);
    int idx = channel->index();
    assert(0 <= idx && idx <static_cast<int>(pollfds_.size()));
    struct pollfd& pfd = pollfds_[idx];
    assert(pfd.fd == channel->fd() || pfd.fd == -1);
    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0;
    if (channel->isNoneEvent()) {
      // ignore this pollfd
      pfd.fd = -1;
    }
  }
}

到目前位置，我们已经将我们关心的描述符及事件都设置好了，现在就需要进行真正的poll调用了，也就是loop循环

void EventLoop::loop()
{
  assert(!looping_);
  assertInLoopThread();
  looping_ = true;
  quit_ = false;

  while (!quit_)
  {
    activeChannels_.clear();
    //在这个poll调用中封装了真正的poll(2)
    poller_->poll(kPollTimeMs, &activeChannels_);
    for (ChannelList::iterator it = activeChannels_.begin();
        it != activeChannels_.end(); ++it)
    {
      (*it)->handleEvent();
    }
  }

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

EventLoop类中保存了一个vector类型的activeChannels，用来获取活跃的事件，因为从上文Poller类中的一个channel变量（map）就可以看出Channel和fd是可以一一对应的。

再来看看Poller封装的poll调用。

Timestamp Poller::poll(int timeoutMs, ChannelList* activeChannels)
{
  // XXX pollfds_ shouldn't change
  int numEvents = ::poll(&*pollfds_.begin(), pollfds_.size(), timeoutMs);//poll调用
  LOG_INFO<  Timestamp now(Timestamp::now());
  if (numEvents > 0) {
    LOG_TRACE <" events happended";
    /*填写活动事件*/
    fillActiveChannels(numEvents, activeChannels);//将活跃的channel添加到avtiveChannel中去
    //并且设置本channel对象的revent
  } else if (numEvents == 0) {
    LOG_TRACE <<" nothing happended";
  } else {
      LOG_INFO<<"time out";
    LOG_SYSERR <<"Poller::poll()";
  }
  return now;
}

最后，看到一个for循环，因为活跃的描述符/Channel可能很多，muduo 的做法，是现将所有活跃的描述符/Channel用activeChannel管理起来再迭代一次，对于每个Channel调用它们的handleEvent方法，该方法就根据设置的revent调用之前设置的回调方法。

void Channel::handleEvent()
{
    /*活动事件回调*/
  if (revents_ & POLLNVAL) {
    LOG_WARN <<"Channel::handle_event() POLLNVAL";
  }

  if (revents_ & (POLLERR | POLLNVAL)) {
    if (errorCallback_) errorCallback_();
  }
  if (revents_ & (POLLIN | POLLPRI | POLLRDHUP)) {
    if (readCallback_) readCallback_();
  }
  if (revents_ & POLLOUT) {
    if (writeCallback_) writeCallback_();
  }
}

4.参考

1.Linux多线程服务端编程 使用MuduoC++网络库
2.http://blog.csdn.net/jnu_simba/article/details/14486661