Reactor模式解析——muduo网络库

最近一段时间阅读了muduo源码,读完的感觉有一个感觉就是有点乱。固然不是说代码乱,是我可能尚未彻底消化和理解。为了更好的学习这个库,仍是要来写一些东西促进一下。linux

我一边读一边尝试在一些地方改用c++11的新特性,这个工做持续在进行中。为啥这么干?没什么理由,纯粹是为了学习。c++

注:本文的大部分代码和图文都来自《Linux多线程服务端编程》,可直接参考muduo的源码,或者参考我这里抄着玩儿的版本。git

Reactor介绍

什么是Reactor? 换个名词“non-blocking IO + IO multiplexing”,意思就显而易见了。Reactor模式用非阻塞IO+poll(epoll)函数来处理并发,程序的基本结构是一个事件循环,以事件驱动和事件回调的方式实现业务逻辑。github

while(!done)
{
    int retval  = poll(fds,nfds,timeout)
    if(retval < 0)
        处理错误,回调用户的error handler
    else{
        处理到期的timers,回调用户的timer handler
        if(retval > 0){
            处理IO事件,毁掉用户的IO event handler
        }
    }
}

这段代码形式上很是简单,跟我上一篇文章epoll的例子十分类似,除了没有处理超时timer部分。在muduo的实现中,定时器使用了linux平台的timerfd_*系列函数, timers和其它IO统一了起来。编程

单线程Reactor实现

muduo的Reactor核心主要由Channel、EventLoop、Poller、TimerQueue这几个类完成。乍一看还有一点绕,代码里面各类回掉函数看起来有点不直观。另外,这几个类的生命周期也值得注意,容易理不清楚。数组

1. Channel

Channel类比较简单,负责IO事件分发,每个Channel对象都对应了一个fd,它的核心成员以下:网络

EventLoop* loop_;
  const int fd_;
  int events_;
  int revents_;
  int index_;

  ReadEventCallback readCallback_;
  EventCallback writeCallback_;
  EventCallback errorCallback_;
  EventCallback closeCallback_;

几个callback函数都是c++新标准里面的function对象(muduo里面是Boost::function),它们会在handleEvent这个成员函数中根据不一样的事件被调用。index_是poller类中pollfds_数组的下标。events_和revents_明显对应了struct pollfd结构中的成员。须要注意的是,Channel并不拥有该fd,它不会在析构函数中去关闭这个fd(fd是由Socket类的析构函数中关闭,即RAII的方法),Channel的生命周期由其owner负责。数据结构

2. Poller

Poller类在这里是poll函数的封装(在muduo源码里面是抽象基类,支持poll和epoll),它有两个核心的数据成员:多线程

typedef std::vector<struct pollfd> PollFdList;
  typedef std::map<int, Channel*> ChannelMap;  // fd to Channel
  PollFdList pollfds_;
  ChannelMap channels_;

ChannelMap是fd到Channel类的映射,PollFdList保存了每个fd所关心的事件,用做参数传递到poll函数中,Channel类里面的index_便是这里的下标。Poller类有下面四个函数并发

Timestamp poll(int timeoutMs, ChannelList* activeChannels);
void updateChannel(Channel* channel);
void removeChannel(Channel* channel);
private:
void fillActiveChannels(int numEvents, ChannelList* activeChannels) const;

updateChannel和removeChannel都是对上面两个数据结构的操做,poll函数是对::poll的封装。私有的fillActiveChannels函数负责把返回的活动时间添加到activeChannels(vector<Channel*>)这个结构中,返回给用户。Poller的职责也很简单,负责IO multiplexing,一个EventLoop有一个Poller,Poller的生命周期和EventLoop同样长。

3. EventLoop

EventLoop类是核心,大多数类都会包含一个EventLoop*的成员,由于全部的事件都会在EventLoop::loop()中经过Channel分发。先来看一下这个loop循环:

while (!quit_)
{
    activeChannels_.clear();
    pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
    for (ChannelList::iterator it = activeChannels_.begin();
        it != activeChannels_.end(); ++it)
    {
      (*it)->handleEvent(pollReturnTime_);
    }
    doPendingFunctors();
}

handleEvent是Channel类的成员函数,它会根据事件的类型去调用不一样的Callback。循环末尾还有一个doPendingFunctors(),这个函数的做用在后面多线程的部分去说明。
由上面三个类已经能够构成Reactor的核心,整个流程以下:

  • 用户经过Channel向Poller类注册fd和关心的事件

  • EventLoop从poll中返回活跃的fd和对应的Channel

  • 经过Channel去回掉相应的时间。

muduo的书里面有一个时序图(8-1),很清楚的说明了整个流程。

4. TimerQueue

muduo的定时器直接使用了标准的容器库set来管理。先看一下TimerQueue:

typedef std::shared_ptr<Timer> TimerPtr;
typedef std::pair<Timestamp, TimerPtr> Entry;
typedef std::set<Entry> TimerList;
Channel timerfdChannel_;
const int timerfd_;
TimerList timers_;

采用std::pair<Timestamp, TimerPtr> 加上set的 的形式是为了处理两个Timer同事到期的状况,即便到期时间相同,它们的地址也不一样。timerfdChannel_是用来管理timerfd_create函数建立的fd。Timer类里面包含了一个回调函数和一个到期时间。expiration_就是上面Entry中的Timestamp。

const TimerCallback callback_;
Timestamp expiration_;

这样整个思路就很清晰了:

  • 用一个set来保存全部的事件和时间

  • 根据set集合里面最先的时间来更新timerfd_的到期时间(用timerfd_settime函数)

  • 时间到期后,EventLoop的poll函数会返回,并调用timerfdChannel_里面的handleEvent回调函数。

  • 经过handleEvent这个回调函数,再去处理到期的全部事件。

timerfdChannel_.setReadCallback(
    std::bind(&TimerQueue::handleRead,this));
timerfdChannel_.enableReading();

timerfdChannel_的callback函数注册了TimerQueue的handleRead函数。在handleRead中应该干什么就很明显了,天然是捞出全部到期的timer,一次去执行对应的事件:

void TimerQueue::handleRead()
{
  loop_->assertInLoopThread();
  Timestamp now(Timestamp::now());
  readTimerfd(timerfd_, now);
  std::vector<Entry> expired = getExpired(now);
  // safe to callback outside critical section
  for (std::vector<Entry>::iterator it = expired.begin();
      it != expired.end(); ++it)
  {
    it->second->run();
  }
  reset(expired, now);
}

至此为止,单线程的Reator就已经完成了。总感受muduo这种事件回调的代码风格,读起来比较绕,不够直观。不知道其余的Reactor模式的网络程序会不会也是这种感受。

多线程的技巧

多线程本质上是困难的,由于它强迫你的大脑去思考两件事情同时发生会出现的各类状况。我目前感受除了看别人的经验和总结,没有什么技巧或者方法论来解决多线程的问题。

一个线程一个EventLoop,每一个线程都有本身管理的各类ChannelList和TimerQueue。有时候,咱们总有一些需求,要在各个线程之间调配任务。好比添加一个定时时间到IO线程中,这样TimerQueue就有两个线程同时访问。我认为muduo在处理上锁的问题上,很值得学习。

1. runInLoop()和queueInLoop()

先来看几个EventLoop里面重要的函数和成员:

std::vector<Functor> pendingFunctors_; // @GuardedBy mutex_
void EventLoop::runInLoop(Functor&& cb) {
  if (isInLoopThread()) {
    cb();
  } else {
    queueInLoop(std::move(cb));
  }
}
void EventLoop::queueInLoop(Functor&& cb) {
  {
    MutexLockGuard lock(mutex_);
    pendingFunctors_.push_back(cb);
  }
  if (!isInLoopThread() || callingPendingFunctors_) {
    wakeup();
  }
}

注意这里的函数参数,我用到了C++11的右值引用。
在前面的EventLoop::loop里面,咱们已经看到了doPendingFunctors()这个函数,EventLoop还有一个重要的成员pendingFunctors_,该成员是暴露给其余线程的。这样,其余线程向IO线程添加定时时间的流程就是:

  • 其余线程调用runInLoop(),

  • 若是不是当前IO线程,再调用queueInLoop()

  • 在queueLoop中,将时间push到pendingFunctors_中,并唤醒当前IO线程
    注意这里的唤醒条件:不是当前IO线程确定要唤醒;此外,若是正在调用Pending functor,也要唤醒;(为何?,由于若是正在执行PendingFunctor里面,若是也执行了queueLoop,若是不唤醒的话,新加的cb就不会当即执行了。)

2.doPendingFunctors()

如今来看一下doPendingFunctors()这个函数:

void EventLoop::doPendingFunctors() {
  std::vector<Functor> functors;
  callingPendingFunctors_ = true;
  {
    // reduce the lenth of the critical section
    // avoid the dead lock cause the functor can call queueInloop(;)
    MutexLockGuard lock(mutex_);
    functors.swap(pendingFunctors_);
  }
  for (size_t i = 0; i < functors.size(); ++i) {
    functors[i]();
  }
  callingPendingFunctors_ = false;
}

doPendingFunctors并无直接在临界区去执行functors,而是利用了一个栈对象,把事件swap到栈对象中,再去执行。这样作有两个好处:

  1. 减小了临界区的长度,其它线程调用queueInLoop对pendingFunctors加锁时,就不会被阻塞

  2. 避免了死锁,可能在functors里面也会调用queueInLoop(),从而形成死锁。

回过头来看,muduo在处理多线程加锁访问共享数据的策略上,有一个很重要的原则:拼命减小临界区的长度
试想一下,若是没有pendingFunctors_这个数据成员,咱们要想往TimerQueue中添加timer,确定要对TimerQueue里面的insert函数加锁,形成锁的争用,而pendingFunctors_这个成员将锁的范围减小到了一个vector的push_back操做上。此外,在doPendingFunctors中,利用一个栈对象减小临界区,也是很巧妙的一个重要技巧。

3.wake()

前面说到唤醒IO线程,EventLoop阻塞在poll函数上,怎么去唤醒及时它?之前的作法是利用pipe,向pipe中写一个字节,监视在这个pipe的读事件的poll函数就会马上返回。在muduo中,采用了linux中eventfd调用

static int createEventfd() {
  int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
  if (evtfd < 0) {
    LOG_SYSERR << "Failed in eventfd";
    abort();
  }
  return evtfd;
}
void EventLoop::wakeup() {
  uint64_t one = 1;
  ssize_t n = ::write(wakeupFd_, &one, sizeof one);
  if (n != sizeof one) {
    LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
  }
}

把eventfd获得的fd和前面同样,经过Channel注册到poll里面,唤醒的时候,只须要向wakeupFd中写入一个字节,就能达到唤醒的目的。eventfd、timerfd都体现了linux的设计哲学,Everyting is a fd。

关于muduod的Reactor模式,如今我终于有一些理解了。关于TCPServer部分,下一篇再写。下一步,我会继续研究muduo的HTTP示例,并尝试扩展它。

相关文章
相关标签/搜索