muduo源码分析：Poller类 (EpollPoller类)

妖狐艹你老母 2022-05-15 19:06 230阅读 0赞

### 2.I/O复用类Poller ###

Poller类是个基类，它的定义如下:

class Poller : boost::noncopyable
        {
            public:
                typedef std::vector<Channel *> ChannelList;
    
                Poller(EventLoop *loop);
    
                virtual ~Poller();
    
                //不许在I/O线程中调用,I/O复用的封装
                virtual Timestamp poll(int timeoutMs,ChannelList *activeChannels) = 0;
    
                //更新Channel
                virtual void updateChannel(Channel *channel) = 0;
    
                //移除Channel
                virtual void removeChannel(Channel *channel) = 0;
    
                //这个channel是否在map中存在
                virtual bool hasChannel(Channel *channel)const;
    
                //默认poller方式
                static Poller *newDefaultPoller(EventLoop *loop);
    
                void assertInLoopThread()const
                {
                    ownerLoop_->assertInLoopThread();
                }
    
            protected:
                typedef std::map<int,Channel*> ChannelMap;    //map关联容器的关键字为channel_所管理的fd
                ChannelMap Channels_;                   //存储事件分发器的map
            private:
                EventLoop *owerLoop_;                   //属于哪个loop
        };

事件分发器channel集用关联容器map来保存，map的关键字为channel所管理的fd。

看下epoll对Poller的实现

EPollPoller类的定义为: EPollPoller.h

class EPollPoller : public Poller
    {
    public:
        EPollPoller(EventLoop* loop);
        virtual ~EPollPoller();
    
        virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels);	//内部调用epoll_wait函数
        virtual void updateChannel(Channel* channel);    //添加更新事件			
        virtual void removeChannel(Channel* channel);    //移除事件
    
    private:
        static const int kInitEventListSize = 16;			//epoll事件表的大小
    
        static const char* operationToString(int op);
    
        void fillActiveChannels(int numEvents,
                              ChannelList* activeChannels) const;	//将epoll返回的活跃事件填充到activeChannel
        void update(int operation, Channel* channel);		//注册事件
    
        typedef std::vector<struct epoll_event> EventList;	//typedef:事件数组
    
        int epollfd_;	        //epoll的事件表fd: epoll例程
        EventList events_;	        //epoll事件数组
    };

Poller主要功能如下：

1.调用poll函数监听注册了事件的文件描述符
    2.当poll返回时将发生事件的事件集装入activeChannels中,并设置Channel发生事件到其revents_中
    3.控制channel中事件的增删改

**EPollPoller.cc完整代码：**

#include <muduo/net/poller/EPollPoller.h>
    
    #include <muduo/base/Logging.h>
    #include <muduo/net/Channel.h>
    
    #include <boost/static_assert.hpp>
    
    #include <assert.h>
    #include <errno.h>
    #include <poll.h>
    #include <sys/epoll.h>
    #include <unistd.h>
    
    using namespace muduo;
    using namespace muduo::net;
    
    
    BOOST_STATIC_ASSERT(EPOLLIN == POLLIN);
    BOOST_STATIC_ASSERT(EPOLLPRI == POLLPRI);
    BOOST_STATIC_ASSERT(EPOLLOUT == POLLOUT);
    BOOST_STATIC_ASSERT(EPOLLRDHUP == POLLRDHUP);
    BOOST_STATIC_ASSERT(EPOLLERR == POLLERR);
    BOOST_STATIC_ASSERT(EPOLLHUP == POLLHUP);
    
    namespace
    {
    const int kNew = -1;
    const int kAdded = 1;
    const int kDeleted = 2;
    }
    
    EPollPoller::EPollPoller(EventLoop* loop)
      : Poller(loop),
        epollfd_(::epoll_create1(EPOLL_CLOEXEC)),
        events_(kInitEventListSize)
    {
      if (epollfd_ < 0)
      {
        LOG_SYSFATAL << "EPollPoller::EPollPoller";
      }
    }
    
    EPollPoller::~EPollPoller()
    {
      ::close(epollfd_);		//关闭事件例程
    }
    
    Timestamp EPollPoller::poll(int timeoutMs, ChannelList* activeChannels)
    {
      LOG_TRACE << "fd total count " << channels_.size();	//channels是一个map<fd, Channel*>，保存了fd对应的channel
      
      //调用epoll_wait监听就绪事件的文件描述符,同时保存发生事件的文件描述符到events_begin()开始的缓冲中
      int numEvents = ::epoll_wait(epollfd_,
                                   &*events_.begin(),
                                   static_cast<int>(events_.size()),
                                   timeoutMs);
      int savedErrno = errno;
      Timestamp now(Timestamp::now());
      if (numEvents > 0)
      {
        LOG_TRACE << numEvents << " events happened";
        fillActiveChannels(numEvents, activeChannels);		//将就绪的channel放到activeChannels，EventLoop统一处理
        if (implicit_cast<size_t>(numEvents) == events_.size())
        {
          events_.resize(events_.size()*2);
        }
      }
      else if (numEvents == 0)
      {
        LOG_TRACE << "nothing happened";
      }
      else
      {
        // error happens, log uncommon ones
        if (savedErrno != EINTR)
        {
          errno = savedErrno;
          LOG_SYSERR << "EPollPoller::poll()";
        }
      }
      return now;
    }
    
    //将就绪事件放到activeChannels中
    void EPollPoller::fillActiveChannels(int numEvents, ChannelList* activeChannels) const
    {
      assert(implicit_cast<size_t>(numEvents) <= events_.size());
      for (int i = 0; i < numEvents; ++i)
      {
    	// epoll_event.data.ptr存放的是Channel的指针
        Channel* channel = static_cast<Channel*>(events_[i].data.ptr);	//在poll()函数中已经调用epoll_wait把发生事件的文件描述符存储到event_中
    #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);	//设置发生的事件到Channel::revents,供Channel::handleEvent使用,以调用相应的回调函数!_
        activeChannels->push_back(channel);			//将就绪事件放入到就绪事件列表中
      }
    }
    
    void EPollPoller::updateChannel(Channel* channel)	//添加Channel到channels_事件分发器map中,并注册Channel对应的事件
    {
      Poller::assertInLoopThread();
      const int index = channel->index();
      LOG_TRACE << "fd = " << channel->fd()
        << " events = " << channel->events() << " index = " << 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
        {
          assert(channels_.find(fd) != channels_.end());
          assert(channels_[fd] == channel);
        }
    
        channel->set_index(kAdded);
        update(EPOLL_CTL_ADD, channel);			//注册事件
      }
      else
      {
        // update existing one with EPOLL_CTL_MOD/DEL
        int fd = channel->fd();
        (void)fd;
        assert(channels_.find(fd) != channels_.end());
        assert(channels_[fd] == channel);
        assert(index == kAdded);
        if (channel->isNoneEvent())
        {
          update(EPOLL_CTL_DEL, channel);		//删除事件
          channel->set_index(kDeleted);
        }
        else
        {
          update(EPOLL_CTL_MOD, channel);
        }
      }
    }
    
    void EPollPoller::removeChannel(Channel* channel) //从事件分发器map channels_中移除Channel,并删除对应的事件
    {
      Poller::assertInLoopThread();
      int fd = channel->fd();
      LOG_TRACE << "fd = " << fd;
      assert(channels_.find(fd) != channels_.end());
      assert(channels_[fd] == channel);
      assert(channel->isNoneEvent());
      int index = channel->index();
      assert(index == kAdded || index == kDeleted);
      size_t n = channels_.erase(fd);	    //移除
      (void)n;
      assert(n == 1);
    
      if (index == kAdded)
      {
        update(EPOLL_CTL_DEL, channel);		//删除事件
      }
      channel->set_index(kNew);
    }
    
    //注册删除事件核心
    void EPollPoller::update(int operation, Channel* channel)    //operation选项指定添加或者删除事件等
    {
      struct epoll_event event;
      bzero(&event, sizeof event);
      event.events = channel->events();
      event.data.ptr = channel;		//注意这里,epoll_event.data.ptr存放的是Channel的指针:供之后的fillActiveChannels使用 
      int fd = channel->fd();
      LOG_TRACE << "epoll_ctl op = " << operationToString(operation)
        << " fd = " << fd << " event = { " << channel->eventsToString() << " }";
      if (::epoll_ctl(epollfd_, operation, fd, &event) < 0)	//核心:向epollfd_注册事件 
      {
        if (operation == EPOLL_CTL_DEL)
        {
          LOG_SYSERR << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd;
        }
        else
        {
          LOG_SYSFATAL << "epoll_ctl op =" << operationToString(operation) << " fd =" << fd;
        }
      }
    }
    
    const char* EPollPoller::operationToString(int op)
    {
      switch (op)
      {
        case EPOLL_CTL_ADD:
          return "ADD";
        case EPOLL_CTL_DEL:
          return "DEL";
        case EPOLL_CTL_MOD:
          return "MOD";
        default:
          assert(false && "ERROR op");
          return "Unknown Operation";
      }
    }