muduo-net 源码分析（一）——EvenLoop

对于事件的处理，一般来说都是使用 one loop pre thread + threadpool 的处理方式，也就是创建一个 I/O 线程用于监听文件描述符的事件，使用线程池来处理事件；每个线程只能创建一个 EvenLoop 对象，而创建 EvenLoop 的线程也就被称为 I/O 线程；

EvenLoop 的源码相当复杂，我们会先讲解 Channel 和 Poller 的源码，对于一些后面用到的代码我们会先省略不说；

这一节的主要思路在于理清 EvenLoop、Poller 和 Channel 的关系；

对于文件套接字以及其对应的监听事件和触发事件都保存在 Channel 中；又因为 Channel 需要在其对应的 EvenLoop 中被管理，所以还会保存一个 EvenLoop 对象；
index 是一个很重要的参数，会在 Poller 中使用到，这里先不说；

EventLoop* loop_;     			//记录当前 channel 所属的 EvenLoop
const int  fd_;       			//套接字
int        events_;   			//关注事件
int        revents_;  			//触发事件
int        index_;    			//
bool       logHup_;

bool eventHandling_;                //是否处于处理事件中
bool addedToLoop_;                  //是否添加到 EvenLoop 中

//三个事件类型
static const int kNoneEvent;      
static const int kReadEvent;
static const int kWriteEvent;

const int Channel::kNoneEvent = 0;
const int Channel::kReadEvent = POLLIN | POLLPRI;   
const int Channel::kWriteEvent = POLLOUT;

//四种事件的函数指针
ReadEventCallback readCallback_;    
EventCallback writeCallback_;
EventCallback closeCallback_;
EventCallback errorCallback_;

//构造函数和析构函数，只传递文件描述符和所属的 EvenLoop 对象
Channel(EventLoop* loop, int fd);
~Channel();

//事件处理函数
void handleEvent(Timestamp receiveTime);
void handleEventWithGuard(Timestamp receiveTime);

//注册、修改和删除事件
//注册和修改最终都是调用 update 函数
void enableReading() { events_ |= kReadEvent; update(); }
void disableReading() { events_ &= ~kReadEvent; update(); }
void enableWriting() { events_ |= kWriteEvent; update(); }
void disableWriting() { events_ &= ~kWriteEvent; update(); }
void disableAll() { events_ = kNoneEvent; update(); }
void remove();
void update();

//设置 Channel 的四种事情发生时调用的函数
typedef std::function EventCallback;      
typedef std::function ReadEventCallback;  
void setReadCallback(ReadEventCallback cb)
{ readCallback_ = std::move(cb); }
void setWriteCallback(EventCallback cb)
{ writeCallback_ = std::move(cb); }
void setCloseCallback(EventCallback cb)
{ closeCallback_ = std::move(cb); }
void setErrorCallback(EventCallback cb)
{ errorCallback_ = std::move(cb); }

//构造函数，注意 index_ 默认是 -1，events_ 默认是 kNoneEvent 即没有任何事件
Channel::Channel(EventLoop* loop, int fd__)
  : loop_(loop),
    fd_(fd__),
    events_(0),
    revents_(0),
    index_(-1),
    logHup_(true),
    tied_(false),
    eventHandling_(false),
    addedToLoop_(false)
{
}

Channel::~Channel()
{
  assert(!eventHandling_);
  assert(!addedToLoop_);
  //析构的条件是管道从 loop_ 中移除
  if (loop_->isInLoopThread())
  {
    assert(!loop_->hasChannel(this));
  }
}

//调用 EvenLoop 的 updateChannel 函数
void Channel::update()
{
  addedToLoop_ = true;
  loop_->updateChannel(this);
}

//调用 EvenLoop 的 removeChannel 函数
//注意在调用 remove 之前一定要先调用 disableAll()，不然断言会失败
void Channel::remove()
{
  assert(isNoneEvent());
  addedToLoop_ = false;
  loop_->removeChannel(this);
}

//handleEvent 实际上就是调用 handleEventWithGuard
void Channel::handleEvent(Timestamp receiveTime)
{
  
  else
  {
    handleEventWithGuard(receiveTime);
  }
}


//handleEventWithGuard() 函数本质上就是根据 revents 的类型来调用发生事件的函数
void Channel::handleEventWithGuard(Timestamp receiveTime)
{
  eventHandling_ = true;
  LOG_TRACE << reventsToString();
  if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
  {
    if (logHup_)
    {
      LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLHUP";
    }
    if (closeCallback_) closeCallback_();
  }

  if (revents_ & POLLNVAL)
  {
    LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLNVAL";
  }

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

具体的函数实现后面会说，大致的过程如下：

EvenLoop ep
Channel ch(&ep, fd);
ch.setReadCallback(std::move(func));
ch.enableReading();
ch.disableAll();
ch.remove();

muduo 中的 Poller 实际上是一个虚基类，实际的实现在 muduo/poller 中；也就是说 muduo 库中主要使用了 poll/epoll 进行事件监听；

EventLoop* ownerLoop_;			//所属的 EvenLoop 对象
typedef std::map ChannelMap;   //int 是文件描述符 
ChannelMap channels_;			//使用 map 的原因是可以通过 key 快速定位 value

//构造函数传递 EvenLoop 对象
Poller(EventLoop* loop)
  : ownerLoop_(loop)
{
}

virtual ~Poller();

typedef std::vector ChannelList;
virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;

virtual void updateChannel(Channel* channel) = 0;

virtual void removeChannel(Channel* channel) = 0;

//该函数只能在所属的 IO 线程中调用
//assertInLoopThread 主要是确定函数是在 IO 线程调用的
//功能就是查找当前 IO 线程是否否在 channel 对象，这也是用 map 的原因
bool hasChannel(Channel* channel) const
{
  assertInLoopThread();
  ChannelMap::const_iterator it = channels_.find(channel->fd());
  return it != channels_.end() && it->second == channel;
}

static Poller* newDefaultPoller(EventLoop* loop);

//根据环境变量生成一个子类对象传递给父类指针
//默认是使用 EPOLL 模式
Poller* Poller::newDefaultPoller(EventLoop* loop)
{
  if (::getenv("MUDUO_USE_POLL"))
  {
    return new PollPoller(loop);
  }
  else
  {
    return new EPollPoller(loop);
  }
}

void assertInLoopThread() const
{
  ownerLoop_->assertInLoopThread();
}

具体的实现先不说，主要是要理清每个类的主要作用；

EvenLoop 中的变量和函数很多，我们分开说；

bool looping_; 						//是不是在调用 poll() 中
std::atomic quit_;      		//是否退出
bool eventHandling_;				//是否在处理事件

//bool callingPendingFunctors_;		本节不讨论		

int64_t iteration_;

const pid_t threadId_;				//创建 EvenLoop 对象的线程 tid            
Timestamp pollReturnTime_;       	//调用 poll() 的时间戳，因为 readbackCall 需要一个时间戳
std::unique_ptr poller_;  	//基类指针，这里使用 unique_ptr，说明 poller 的生命周期受 EvenLoop 管理

typedef std::vector ChannelList;
ChannelList activeChannels_;   		//EvenLoop 中活跃的通道
Channel* currentActiveChannel_;		//现在处理的通道

• 本节中最主要的变量就是 threadId_、poller_ 和 activeChannels_；
• 一个线程只能有一个 EvenLoop，一个 EvenLoop 只能有一个 Poller，但一个 EvenLoop 可以有多个 Channel；

//本节只讨论一下函数，有一些返回参数的函数自己看看就行了
EventLoop();
~EventLoop();
void loop();
void quit();
void updateChannel(Channel* channel);       
void removeChannel(Channel* channel);       
bool hasChannel(Channel* channel);

//assertInLoopThread 函数是一个经常使用的函数，因为大部分的 EvenLoop 操作都需要在当前线程中使用
void assertInLoopThread()
{
  if (!isInLoopThread())
  {
    abortNotInLoopThread();
  }
}

//判断当前线程是不是 IO 线程
bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }

__thread EventLoop* t_loopInThisThread = 0;   //线程的局部变量

//EvenLoop 主要的作用就是判断当前线程有没有创建过 EvenLoop 对象
EventLoop::EventLoop()
  : looping_(false),
    quit_(false),
    eventHandling_(false),
    //callingPendingFunctors_(false),
    iteration_(0),
    threadId_(CurrentThread::tid()),		//获取创建 EvenLoop 对象的线程 tid
    poller_(Poller::newDefaultPoller(this)),	//这里是返回一个派生类指针，面向对象的思路
    //timerQueue_(new TimerQueue(this)),
    //wakeupFd_(createEventfd()),
    //wakeupChannel_(new Channel(this, wakeupFd_)),
    currentActiveChannel_(NULL)
{
  LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
  //判断当前线程是否已经存在 EvenLoop，如果已经存在则报错
  //一开始 t_loopInThisThread 为 0
  if (t_loopInThisThread)
  {
    LOG_FATAL << "Another EventLoop " << t_loopInThisThread
              << " exists in this thread " << threadId_;
  }
  else
  {
    t_loopInThisThread = this;
  }
  
  
}


EventLoop::~EventLoop()
{
  LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
            << " destructs in thread " << CurrentThread::tid();
  
  t_loopInThisThread = NULL;
}


//重点函数之一，只能在 IO 线程中调用
void EventLoop::loop()
{
  assert(!looping_);
  assertInLoopThread();
  looping_ = true;
  quit_ = false;  
  LOG_TRACE << "EventLoop " << this << " start looping";

  //循环条件，不退出
  while (!quit_)
  {
    //清空活跃列表
    activeChannels_.clear();
    
    //实际上 EvenLoop 的 poll() 函数就是调用 Poller 派生类的 poll() 函数
 	//在这里我们认为 poll() 函数是返回调用 poll() 的时间戳，以及返回 activeChannels_
	pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
    
    ++iteration_;

	//输出日志
    if (Logger::logLevel() <= Logger::TRACE)
    {
      printActiveChannels();
    }

    eventHandling_ = true;

    //遍历所有活跃事件，调用活跃事件的 handleEvent 函数
    //前面说过 Channel 的 handleEvent() 是根据活跃通道的 revent 来调用对应函数
    for (Channel* channel : activeChannels_)
    {
      currentActiveChannel_ = channel;
      currentActiveChannel_->handleEvent(pollReturnTime_);
    }
    
    //事件处理结束，当前没有活跃事件
    currentActiveChannel_ = NULL;
    eventHandling_ = false;
    
    //doPendingFunctors(); 		本节不解释
  }

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

//活跃事件处理完后退出循环
void EventLoop::quit()
{
  quit_ = true;
  
}

//只能在 IO 线程中调用，实际上调用的是 Poller 的 updateChannel
//Channel 的 update 调用的是 EvenLoop 的 updateChannel，但实际都是调用 Poller 的 updateChannel
void EventLoop::updateChannel(Channel* channel)
{
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  poller_->updateChannel(channel);
}

//只能在 IO 线程中调用，实际上调用的是 Poller 的 removeChannel
//Channel 的 update 调用的是 EvenLoop 的 removeChannel，但实际都是调用 Poller 的 removeChannel
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);
}

bool EventLoop::hasChannel(Channel* channel)
{
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  return poller_->hasChannel(channel);
}

void EventLoop::abortNotInLoopThread()
{
  LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
            << " was created in threadId_ = " << threadId_
            << ", current thread id = " <<  CurrentThread::tid();
}

根据现在的函数和变量总结一下 Channel、Poller 和 EvenLoop 的关系；

1. 首先在当前线程创建一个 EvenLoop 对象，这里会默认创建一个 Poller 对象；
2. 创建多个 Channel 对象，传递 EvenLoop 对象以及他们的 fd；
3. 设置 Channel 的事件函数，并设置 Channel 对象的监听事件，enableReading()、enableWriting()…，这些函数会自动调用 update() 函数，最终调用 Poller 的 updateChannel() 函数；
4. 开启循环，调用 EvenLoop 的 poll()，实际调用的是 Poller 中的 poll() 函数，然后处理活跃事件的任务；
5. 关闭 Channel 的监听事件，调用 remove() ，实际上调用的是 Poller 的 removeChannel() 函数；
6. 退出循环，调用 EvenLoop的 quit()；

下一节需要处理的 Poller 的派生类，主要是 poll()、updateChannel() 和 removeChannel()；我们主要以 poll 作为讲解，epoll 和 poll 差不多，对于不同的地方会做出讲解；