muduo网络库——poller

前言

muduo网络库的多路复用是封装的poll和epoll，而基类poller提供了统一的接口。

关于select/poll/epoll的介绍，可以参考：linux进阶09——I/O（二）：Select、Poll和Epoll_却道天凉_好个秋的博客-CSDN博客​​​​​​

源码分析 Poller

class Poller : noncopyable
{
 public:
  typedef std::vector ChannelList;

  Poller(EventLoop* loop);
  virtual ~Poller();

  // 处理I/O事件，必须在EventLoop所在的线程中被调用
  virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;

  // 更新通道，一个通道对应一个文件描述符(fd), 文件描述符用来监听读/写事件,必须在EventLoop所在的线程中被调用
  virtual void updateChannel(Channel* channel) = 0;

  // 移除通道接口，必须在EventLoop所在的线程中被调用
  virtual void removeChannel(Channel* channel) = 0;

  // 判断当前Poller对象是否持有指定通道
  virtual bool hasChannel(Channel* channel) const;

  // 默认创建Poller对象的类函数
  static Poller* newDefaultPoller(EventLoop* loop);

  // 判断是否属于当前EventLoop线程
  void assertInLoopThread() const
  {
    ownerLoop_->assertInLoopThread();
  }

 protected:
  typedef std::map ChannelMap;
  ChannelMap channels_;

 private:
  EventLoop* ownerLoop_;  // 事件循环
};

EPollPoller EPollPoller

EPollPoller构造函数中，通过epoll_create1创建epoll实例, epoll_create1相比较epoll_create,具有具有执行后关闭的特性。

EPOLL_CLOEXEC
Set the close-on-exec (FD_CLOEXEC) flag on the new file descrip‐
tor.  See the description of the O_CLOEXEC flag in  open(2)  for
reasons why this may be useful

构造函数源码如下： 

EPollPoller::EPollPoller(EventLoop* loop)
  : Poller(loop),
    epollfd_(::epoll_create1(EPOLL_CLOEXEC)),    // 通过epoll_create1创建epoll实例, epoll_create1相比较epoll_create,具有具有执行后关闭的特性
    events_(kInitEventListSize)
{
  if (epollfd_ < 0)
  {
    LOG_SYSFATAL << "EPollPoller::EPollPoller";
  }
}

poll

poll函数中主要通过调用epoll_wait函数将激活的fd添加到激活队列activeChannels中。

Timestamp EPollPoller::poll(int timeoutMs, ChannelList* activeChannels)  
{
  LOG_TRACE << "fd total count " << channels_.size();
  int numEvents = ::epoll_wait(epollfd_,
                               &*events_.begin(),
                               static_cast(events_.size()),
                               timeoutMs);    // 通过epoll_wait等待事件发生
  int savedErrno = errno;
  Timestamp now(Timestamp::now());
  if (numEvents > 0)  // 有事件发生
  {
    LOG_TRACE << numEvents << " events happened";
    fillActiveChannels(numEvents, activeChannels);
    if (implicit_cast(numEvents) == events_.size())
    {
      events_.resize(events_.size()*2);
    }
  }
  ...  

  return now;
}

void EPollPoller::fillActiveChannels(int numEvents,
                                     ChannelList* activeChannels) const
{
  assert(implicit_cast(numEvents) <= events_.size());
  for (int i = 0; i < numEvents; ++i)
  {
    Channel* channel = static_cast(events_[i].data.ptr);
#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);
  }
}

poll函数是跑在事件循环（EventLoop）中，EventLoop中有一个线程进行事件查询，当检测到激活队列中有fd时，通过handeEvent函数处理，具体如下：

void EventLoop::loop()
{
  assert(!looping_);
  assertInLoopThread();
  looping_ = true;
  quit_ = false;  // FIXME: what if someone calls quit() before loop() ?
  LOG_TRACE << "EventLoop " << this << " start looping";

  while (!quit_)  // 只有事件循环quit时，quit_ = false 循环才会退出
  {
    activeChannels_.clear();
    pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);   //调用poll接口获取事件
    ++iteration_;
    if (Logger::logLevel() <= Logger::TRACE)
    {
      printActiveChannels();
    }
    // TODO sort channel by priority
    eventHandling_ = true;
    for (Channel* channel : activeChannels_)
    {
      currentActiveChannel_ = channel;
      currentActiveChannel_->handleEvent(pollReturnTime_);  //处理获取到的事件
    }
    currentActiveChannel_ = NULL;
    eventHandling_ = false;
    doPendingFunctors();
  }

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

update

update函数主要通过调用epoll_ctl函数来执行修改（EPOLL_CTL_MOD）或删除（EPOLL_CTL_DEL） *** 作。

通过updateChannel和removeChannel两个函数供上层调用。

void EPollPoller::update(int operation, Channel* channel)
{
  struct epoll_event event;
  memZero(&event, sizeof event);
  event.events = channel->events();
  event.data.ptr = channel;
  int fd = channel->fd();
  LOG_TRACE << "epoll_ctl op = " << operationToString(operation)
    << " fd = " << fd << " event = { " << channel->eventsToString() << " }";
  if (::epoll_ctl(epollfd_, operation, fd, &event) < 0)
  {
    if (operation == EPOLL_CTL_DEL)
    {
      LOG_SYSERR << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd;
    }
    else
    {
      LOG_SYSFATAL << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd;
    }
  }
}