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

你的名字 2022-07-14 03:47 265阅读 0赞

作者一直强调的一个概念叫做`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][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 " << threadId_;
      /*t_loopInThisThread不为空表示当前线程创建了EventLoop对象*/
      /*one thread per loop表示一个线程只能有一个EventLoop对象*/
      if (t_loopInThisThread)
      {
    
        LOG_FATAL << "Another EventLoop " << t_loopInThisThread
                  << " exists in this thread " << threadId_;
      }
      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 = " << channel->fd() << " events = " << channel->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<Channel*>`类型的`activeChannels`，用来获取活跃的事件，因为从上文Poller类中的一个channel变量（`map<int ,channel*>`）就可以看出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<<numEvents;
      Timestamp now(Timestamp::now());
      if (numEvents > 0) {
        LOG_TRACE << numEvents << " 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][http_blog.csdn.net_jnu_simba_article_details_14486661]

[https_github.com_chenshuo_recipes]: https://github.com/chenshuo/recipes
[http_blog.csdn.net_jnu_simba_article_details_14486661]: http://blog.csdn.net/jnu_simba/article/details/14486661