muduo网络库源码解析(1):多线程异步日志库(上)
muduo网络库源码解析(2):多线程异步日志库(中)
muduo网络库源码解析(3):多线程异步日志库(下)
muduo网络库源码解析(4):TimerQueue定时机制
muduo网络库源码解析(5):EventLoop,Channel与事件分发机制
muduo网络库源码解析(6):TcpServer与TcpConnection(上)
muduo网络库源码解析(7):TcpServer与TcpConnection(下)
muduo网络库源码解析(8):EventLoopThreadPool与EventLoopThread
muduo网络库源码解析(9):Connector与TcpClient

引言

这一篇分析muduo的事件分发机制,核心类为EventLoop与Channel,我认为理解的重点在于先不想多线程,就把Eventloop和Channel当做一个单线程的组成部分,这样能够更好的理解代码.但是这一章的很多东西并没有办法讲清楚,因为Eventloop的设计过程我觉得是一步一步往里面加功能的,不太可能一次所有的功能都加进去,解析如果一次全都说了也对理解没有帮助.必须得放到后面几篇分析,所以有些地方适当提一提,只说是什么功能,毕竟这一章我们的重点是事件分发机制,把这个说明白已经很好了.

我们先不去看EventLoop的具体成员,而来想一想这东西到底是干什么的,首先提前说明每一个Eventloop对象都是一个reactor,也就是说每一个EventLoop必然都存在与子线程,那么我们在主线程如何在接到事件连接后把accept到的fd传递给子线程呢,muduo中的做法是把线程所属的eventloop对象的指针传递过去,然后注册回调.这样解释的话我们就清楚eventloop实际上是什么了,实际上就是一个manger类,它管理子线程内的连接,即channel对象的集合以及IO multiplexing对象,那它也一定有一个进行事件循环的函数,来进行正常的事件循环.最有意思的就是主线程的回调如何传递给子线程的Eventloop对象呢,假设你向IO multiplexing加入的时候其正在wait呢?muduo的解决方法是注册一个eventfd,将这个eventfd注册在IO multiplexing中,然后注册一个回调队列,每次由主线程加入时加入到队列中,随后向eventfd写数据,即触发可读事件,在事件循环中取出回调事件执行,这样就完成了一整套操作.

我们首先来看看Eventloop的的构造函数

EventLoop::EventLoop()
  : looping_(false), //正在事件循环时为true,退出事件循环时为false
    quit_(false),  //事件循环的终止标识符
    eventHandling_(false),
    callingPendingFunctors_(false),//协助queueInLoop
    iteration_(0),
    threadId_(CurrentThread::tid()),
    poller_(Poller::newDefaultPoller(this)), //IO multiplexing
    timerQueue_(new TimerQueue(this)), //定时器
    wakeupFd_(createEventfd()), //eventfd 唤醒IO multiplexing
    wakeupChannel_(new Channel(this, wakeupFd_)),  //协助queueInLoop 个人认为的绝对核心 回调通知机制
    currentActiveChannel_(NULL)
{
  LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
  if (t_loopInThisThread)
  {
    LOG_FATAL << "Another EventLoop " << t_loopInThisThread
              << " exists in this thread " << threadId_;
  }
  else
  {
    t_loopInThisThread = this;
  }
  wakeupChannel_->setReadCallback(
      std::bind(&EventLoop::handleRead, this));
  // we are always reading the wakeupfd
  wakeupChannel_->enableReading();
}

成员初始化就不提了,我们来看看函数体部分,t_loopInThisThread是什么,这实际上是一个__Thread的对象

__thread EventLoop* t_loopInThisThread = 0;

这可以保证每个线程只持有一个Eventloop对象,然后我们可以看到wakeupChannel_注册了回调,它的类是channel,我们一起来看看这个类.同样,我们首先来看看构造函数.

Channel::Channel(EventLoop* loop, int fd__)
  : loop_(loop), //记录所属的loop,显然每个channel只能属于一个loop
    fd_(fd__), //每一个channel对应一个fd,但不负责关闭
    events_(0), //关注的事件类型
    revents_(0), //从IOmultiplexing中接收到的事件类型
    index_(-1), //在poll IOmultiplexing中的序号
    logHup_(true),
    tied_(false),//是否绑定
    eventHandling_(false), //是否处于事件循环中
    addedToLoop_(false)
{
}

我们还可以在类的定义中看到这些

  ReadEventCallback readCallback_;
  EventCallback writeCallback_;
  EventCallback closeCallback_;
  EventCallback errorCallback_;
  ...........................
  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(); }
  bool isWriting() const { return events_ & kWriteEvent; }
  bool isReading() const { return events_ & kReadEvent; }
  ........................
  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); }

即一个channel注册的回调,我们看到在注册事件中有一个update.

void Channel::update()
{
  addedToLoop_ = true;
  loop_->updateChannel(this);
}

其会调用所属Eventloop的updateChannel,

void EventLoop::updateChannel(Channel* channel)
{
  assert(channel->ownerLoop() == this);
  assertInLoopThread();
  poller_->updateChannel(channel);
}

Eventloop的updateChannel则会把channel加入IO multiplexing中.这意味着对于每一个channel对象我们在设置事件类型时实际已经把其加入IO multiplexing中.

当然有update也一定有remove,逻辑与上相同

void Channel::remove()
{
  assert(isNoneEvent());
  addedToLoop_ = false;
  loop_->removeChannel(this);
}

我们在来看看channel如何执行回调

void Channel::handleEvent(Timestamp receiveTime) //防止对象执行时被析构 
{
  std::shared_ptr<void> guard;
  if (tied_)
  {
    //最最重要的一句, 在绑定的TcpConnection对象存在时gaurd为true,并延长TcpConnection的生命周期 第六篇中详解
    guard = tie_.lock();
    if (guard)
    {
      handleEventWithGuard(receiveTime); //正常的执行逻辑 调用注册的回调
    }
  }
  else
  {
    handleEventWithGuard(receiveTime);
  } 
}

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;
}

我们可以看到handleEvent中利用了tie_.lock,这个函数的作用如果weak_ptr对应的shared_ptr引用计数不为0的话引用技术加1,否则返回空指针.这一语句是为了防止在回调函数加入线程池时时间循环删除对象,导致函数运行时对象析构,会在第六篇中详解.我们主要看看handleEventWithGuard,就是检测相应的时间,执行相应的回调.

channel便分析完了.

我们继续看Eventloop这个类.我们来看看其事件循环,这当然也是reactor比较重要的地方

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_)
  {
    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_ = channel;
      currentActiveChannel_->handleEvent(pollReturnTime_); //执行每一个事件的回调
    }
    currentActiveChannel_ = NULL;
    eventHandling_ = false;
    doPendingFunctors();//执行在其他线程加入的回调
  }

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

其实就是很正常的事件处理,首先得到事件,然后对每一个事件执行响应的回调.有两个地方比较有意思,值得一说.

首先是currentActiveChannel_与eventHandling_的使用,为什么要用true,false的赋值包起来回调的执行,原因在上一篇Timerqueue中提到过一个类似的写法,原因是为了防止回调中执行removeChannel,显然currentActiveChannel_会在每次执行回调的时候被赋值为当前执行回调的channel.

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);
}

我们可以看到在删除的时候会检测currentActiveChannel_是否为要删除的channel,如果是的话显然就不能进行删除了,没找到的话当然也不能删除.一般情况下不会出现这种错误,至少本线程不会,除非回调被扔到线程池中,才有可能发生这样的事情.出现的时候 退出程序也许是最好的选择.

还有一个地方值得一提,就是doPendingFunctors,这是主线程与子线程通信的渠道.

void EventLoop::doPendingFunctors()
{
  std::vector<Functor> functors;
  callingPendingFunctors_ = true;

  {
  MutexLockGuard lock(mutex_);
  functors.swap(pendingFunctors_); //减小锁的粒度
  }

  for (const Functor& functor : functors)
  {
    functor();
  }
  callingPendingFunctors_ = false; 
  //true false包起来的写法是为了防止回调中调用queueInLoop,否则新加入的回调可能永远不会被调用
}

我们可以看到执行时会在==pendingFunctors_==中执行回调,pendingFunctors_是什么,这是我们会在一个向其中加入回调.

void EventLoop::queueInLoop(Functor cb)
{
  {
  MutexLockGuard lock(mutex_); //显然是需要加锁的,每一个reactor都有一个,可以并发进行,且锁的粒度很小,只是交换指针
  pendingFunctors_.push_back(std::move(cb));
  }

  //不在当前IO线程 但是能跑到这里 调用的wakeup其实就是与本线程不是一个的那个
  if (!isInLoopThread() || callingPendingFunctors_) //见doPendingFunctors
  {
    wakeup(); //向eventfd写入数据,唤醒IO multiplexing
  }
}
...................
void EventLoop::wakeup()
{
  uint64_t one = 1;
  ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
  }
}

queueInLoop怎么用呢,其实就是主线程用来给子线程传递回调的,我们可以看到在queueInLoop中执行wakeup有两个条件,一个是不在当前线程,那当然要调用wakeup来使得注册在reactor的IO multiplexing的eventfd可读,从而唤醒IO multiplexing,执行回调.还有一个呢,callingPendingFunctors_为true时,这与上一篇和这一篇提到的问题类似,就是怕执行回调时执行queueinloop,这样的话万一后面没有事件到来的话,这些回调便永远不能触发了.再回到doPendingFunctors,在执行时把其中的回调都执行一遍,没有什么问题.

说道这,在构造函数中注册的回调大家也应该就明白了

  wakeupChannel_->setReadCallback(
      std::bind(&EventLoop::handleRead, this));
  // we are always reading the wakeupfd
  wakeupChannel_->enableReading();
  ............................
  void EventLoop::handleRead()
{
  uint64_t one = 1;
  ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
  }
}

其实就是给eventfd注册一个可读事件,以支持上述的机制.

总结

这篇文章中主要还是描述了一个问题,即文章开头的那一大段话所陈述的那些,简单来说就是事件分发机制的雏形.其中涉及了一些以前写网络编程代码不太容易注意到的小问题,比如回调的执行可能会遇到对象被析构的问题,这对于编写一个健壮的代码来说是不可或缺的也是不容易想到的,但是相比于这种事件分发机制,我有一种相比之下稍优的结构,下面的简单的说下.包装好的代码在这里,进入链接后channel.h和channel.cc中描述了我的想法.

就是在每一个reactor线程中维护一个queue< int>和eventfd,然后把这两个传到主线程中,每次主线程accept以后选择一个线程,在向队列中加入fd的时候是evenfd加1,然后IO multiplexing被唤醒时只需要在queue取evenfd的值项数的值,这样就不必加锁了.

下面是一个简单的实现,写的时候没太注意代码的规范,抱着轻松的心态看即可.

#include <bits/stdc++.h>
using namespace std;
#include <unistd.h>
#include <sys/eventfd.h>
#include <assert.h>

class EventLoop{
private:
    bool looping;
    const std::thread::id threadID;
    int x;
public:
    EventLoop();
    ~EventLoop();
    void loop();
};

thread_local EventLoop* EventLoopInThisThread = nullptr;

EventLoop::EventLoop()
        : threadID(std::this_thread::get_id()), looping(false), x(5){
    if(EventLoopInThisThread){
        std::cout << "errno in eventloop.cc : \n";
        abort();
    } else {
        EventLoopInThisThread = this;
    }
}

EventLoop::~EventLoop(){
    assert(!looping);
    EventLoopInThisThread = nullptr;
}

void
EventLoop::loop(){
    assert(!looping);
    looping = true;

    //do something.

    looping = false;
}

class reactor{
private:
    EventLoop* event;
    int a= 5;
    std::queue<int> ptr_que;
    int Eventfd_ ;
public:
    reactor(EventLoop* loop, int fd) : event(EventLoopInThisThread), Eventfd_(fd){}

    int show(){
        return a;  
    }
    void set(int x){
        a = x;
    }
    std::queue<int>* return_ptr(){
        return &ptr_que;
    }

    int Return_fd() const{
        return Eventfd_;
    }
};

void test(std::promise<std::queue<int>*>& pro, int T){
    try{
        EventLoop loop;
        reactor rea(&loop, T);
        pro.set_value(rea.return_ptr());
/*         try{
            cout << "ok\n";
            }catch(...){
                pro.set_exception(std::current_exception());
            } */

            while(true){
                sleep(1);
                uint64_t Temp = 0;
                read(rea.Return_fd(), &Temp, sizeof(Temp));
                cout << "size : " << Temp << endl;
                while(Temp--){
                    assert(!rea.return_ptr()->empty());
                    cout << std::this_thread::get_id() << " : " <<rea.return_ptr()->front() << endl;
                    rea.return_ptr()->pop();
                }
            }
    }catch(...){
        std::cerr << "error in : " << std::this_thread::get_id() << std::endl;//log_fatal
    }
}

int main(){
    std::vector<std::thread> pool;
    std::vector<std::future<std::queue<int>*> > vec;
    std::vector<std::queue<int>*> store_;
    std::vector<int> eventfd_;

    for(size_t i = 0; i < 3; i++){
        std::promise<std::queue<int>*> Temp;
        vec.push_back(Temp.get_future());
        int T = eventfd(0,EFD_CLOEXEC | EFD_NONBLOCK);
        pool.push_back(std::thread(test, std::ref(Temp), T));
        store_.push_back(vec[i].get());
        cout << store_[i]->size() << endl;
        eventfd_.push_back(T);
    }
    constexpr uint64_t Temp = 1;
    for(size_t i = 0; i < 3; i++){
        cout << "loop\n";
        store_[i]->push(5);
        write(eventfd_[i], &Temp, sizeof(Temp));
                cout << "two\n";
    }
    sleep(3);
    for(size_t i = 0; i < 3; i++){
        cout << "loop\n";
        store_[i]->push(5);
        write(eventfd_[i], &Temp, sizeof(Temp));
                cout << "two\n";
    }
    std::for_each(pool.begin(), pool.end(), std::mem_fn(&std::thread::join));
    return 0;
}