EventLoop::runInLoop 函数功能

在上一篇博客介绍了 muduo 的核心主循环EventLoop::loop函数, 在 muduo 中, 还有一个十分好用的功能: 可以执行其他线程的任务, 因为平时 IO 线程都阻塞在EventLoop::loop函数的poll函数中, 为了让空闲的 IO 线程也能利用起来, 某一个(IO线程或者其他)线程可以执行一个任务调用EventLoop::runInLoop这个函数, 判断如果当前线程是不是 IO 线程, 如果是就直接执行任务, 不是就添加进任务队列, 并唤醒 IO 线程, 让他执行任务

因为在唤醒 IO 线程时, 用到了 eventfd, 先介绍一下 eventfd

eventfd 的使用

eventfd 是linux 2.6.22后系统提供的一个轻量级的进程间通信的系统调用, 可以进行多进程, 多线程之间的事件通知, eventfd 通过一个进程间共享的64位计数器完成进程间通信，这个计数器由在 linux 内核空间维护，用户可以通过调用 write 方法向内核空间写入一个64位的值，也可以调用 read 方法读取这个值。
eventfd实现了线程之间事件通知的方式, 也可以用于用户态和内核通信. eventfd的缓冲区大小是sizeof(uint64_t); 向其 write 可以递增这个计数器, read操作可以读取, 并进行清零; eventfd也可以放到监听队列中,当计数器不是0时,有可读事件发生,可以进行读取.

三种新的fd都可以进行监听，当有事件触发时，有可读事件发生。
三种新的fd加入linux内核的的版本：
signalfd：2.6.22
timerfd： 2.6.25
eventfd：2.6.22

创建

#include <sys/eventfd.h>
int eventfd ( unsigned int initval, int flags );
//成功返回事件驱动的文件描述符, 失败返回-1

第一个参数用来初始化计数器, 一般设置为0
第二个参数可以使用三个如下三个宏:
- EFD_CLOEXEC : FD_CLOEXEC，简单说就是fork子进程时不继承
- EFD_NONBLOCK : 文件会被设置成O_NONBLOCK，一般要设置
- EFD_SEMAPHORE : (2.6.30以后支持)支持semophore语义的read, 简单说就值递减1

在write之后没有read, 但是又write新的数据, 那么读取的是这两次的8个字节的和

eventfd打开, 读写和关闭都效非常高, 因为它本质并不是文件, 而是kernel在内核空间(内存中)维护的一个64位计数器而已.

这个新建的fd的操作很简单:

read(): 读操作就是将计数器值置0如果是 flag 设置为 semophore 就减1
write(): 设置计数器的值

eventfd 示例代码

#include <sys/eventfd.h>
#include <assert.h>
#include <unistd.h>

#include <iostream>
#include <thread>
using namespace std;

void func(int evfd) 
{
    uint64_t buffer;
    int res;
    while (1) 
    {
        res = read(evfd, &buffer, sizeof(uint64_t));
        assert(res == 8);

        printf("buffer = %lu\n", buffer);
    }
}
int main() 
{
    int evfd = eventfd(1, 1);
    assert(evfd != -1);

    thread t(func, evfd);
    t.detach();

    int res;
    uint64_t buf = 1;
    while (1) 
    {
        res = write(evfd, &buf,sizeof(uint64_t));
        assert(res == 8);
        printf("write = %d\n", res);

        sleep(1);
    }
}

代码运行结果:

lzj@lzj:~/High_per_ser_pro$ ./a.out
write = 8
buffer = 1
buffer = 1
write = 8
buffer = 1
write = 8
buffer = 1
write = 8
buffer = 1
write = 8
buffer = 1
...

EventLoop::runInLoop源码分析

EventLoop所属 IO 线程中 wakeupFd_ 的初始化

//初始化函数
//其他部分省略
EventLoop::EventLoop()
  :  wakeupFd_(createEventfd()),
  wakeupChannel_(new Channel(this, wakeupFd_))
  {
  //...
   //绑定处理wakeup函数写入的数据
  wakeupChannel_->setReadCallback(
      std::bind(&EventLoop::handleRead, this));
  // we are always reading the wakeupfd
  //设置事件类型, 并添加到事件列表
  wakeupChannel_->enableReading();
  }

某一个函数调用EventLoop::runInLoop

void EventLoop::runInLoop(Functor cb) //这个函数可以跨线程调用
{
  if (isInLoopThread())//如果是IO线程立即执行cb()
  {
    cb();
  }
  else//否则放到IO线程的任务队列,也就是std::vector<Functor> pendingFunctors_
  {
    queueInLoop(std::move(cb));
  }
}

void EventLoop::queueInLoop(Functor cb)
{
  {
  MutexLockGuard lock(mutex_);
  pendingFunctors_.push_back(std::move(cb)); //添加进任务队列,异步执行
  }
  
  //唤醒条件
  //不是IO线程
  //是IO线程但是正在执行任务队列中的任务
  if (!isInLoopThread() || callingPendingFunctors_)
  {
    wakeup();//唤醒IO线程
  }
}

唤醒 IO 线程

//唤醒eventloop, 写入8字节数据,为了在需要poll返回时触发事件
void EventLoop::wakeup()
{
  uint64_t one = 1;
  //写入了8字节,在handleRead函数中简单读取了一下,是为了poll返回,也就是要weakup poll
 //wakeupFd_就是eventfd
  ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);  
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
  }
}

还记得EventLoop::loop函数么, 在他的循环中:

while (!quit_)
  {
    activeChannels_.clear(); 
    pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); 
    ++iteration_; 
    if (Logger::logLevel() <= Logger::TRACE)
    {
      printActiveChannels();
    }
    // TODO sort channel by priority
    eventHandling_ = true;
    for (Channel* channel : activeChannels_)
    {
      currentActiveChannel_->handleEvent(pollReturnTime_);  
    }
    currentActiveChannel_ = NULL;
    eventHandling_ = false;
    doPendingFunctors();//执行任务队列中的任务
  }

在处理完就绪事件后, 就会调用EventLoop:: doPendingFunctors()执行任务队列里的任务

// 该函数只会被当前IO线程调用
void EventLoop::doPendingFunctors()
{
  std::vector<Functor> functors;
  callingPendingFunctors_ = true;
 
  {
  MutexLockGuard lock(mutex_);
  functors.swap(pendingFunctors_);
  }
 
  for (size_t i = 0; i < functors.size(); ++i)
  {
    functors[i]();
  }
  callingPendingFunctors_ = false;
}

如何使用该功能

我一开始没有搞懂这个功能的具体原理和调用过程, 看了后面的EventLoopThread才明白

任何一个线程只要创建并运行了EventLoop,就是一个所谓的IO线程, 如果我在主线程里调用EventLoopThread创建一个子线程, 在子线程中创建一个EventLoop对象, 并在主线程中拿到了该EventLoop对象的指针, 然后在主线程调用EventLoop::runInLoop, 那么就不是在IO线程中, 就会调用EventLoop::queueInLoop添加任务到IO线程的任务队列, 然后唤醒IO线程, 执行任务