muduo

Table of Contents

http://code.google.com/p/muduo/

在分析muduo之前必须了解一下作者的想法:http://www.cnblogs.com/Solstice/archive/2010/08/29/muduo_net_lib.html

muduo使用了很多新的Linux内核特性,包括使用signalfd和timerfd来触发信号以及定时器,所以代码上相对于于hpserver好看很多了。但是里面使用了boost::bind以及boost一些东西, 因为个人对于这个部分不是很清楚,所以很多地方并不是非常理解。muduo和hpserver一样也引入了很多概念,了解这些概念也非常有帮助。muduo做了线程管理,但是仅仅做了event loop 的线程管理,没有做工作线程的管理。所以工作线程还是需要自己管理。异步队列在base目录下面也实现了。所以基本上可以认为muduo里面包含了很多网络编程框架需要的组件。 muduo将hpserver下面的event item和event handler以及handle都在一起,称之为Channel.而Reactor在这里称为EventLoop.所以可以认为相对于hpserver,类层次结构好理解多了。

1 base

base下面都是一些关于多线程编程方面需要使用的组件 http://www.cnblogs.com/Solstice/archive/2010/08/21/muduo_thread_lib.html 包括下面这些文件:

  • Atomic.h // 原子操作,里面的CAS没有使用汇编而是使用__sync_val_compare_and_swap这个GCC内置函数
  • BlockingQueue.h // 异步队列,底层使用std::deque来实现,没有大小限制
  • BoundedBlockingQueue.h // 异步队列,但是使用循环数组来实现,有大小限制
  • Condition.h // pthread_cond封装
  • CountDownLatch.h // 可以用作类似于起跑线机制,值得学习一下
void CountDownLatch::wait() // latchdown之后然后调用wait.等待最后一个线程notifyAll,
// 然后多个线程同时解除锁定就可以同时开始执行了
{
    MutexLockGuard lock(mutex_);
    while (count_ > 0) {
        condition_.wait();
    }
}

void CountDownLatch::countDown() // 每个线程在开始都latchdown,最后一个线程会notifyAll
{
    MutexLockGuard lock(mutex_);
    --count_;
    if (count_ == 0) {
        condition_.notifyAll();
    }
}
  • Logging.h // 日志
  • Mutex.h // 互斥锁
  • ProcessInfo.h // 进程信息
  • Singleton.h // 单例模式,实现上比较有特色
static T& instance()
{
  pthread_once(&ponce_, &Singleton::init); // 使用pthread_once来进行构造
  return *value_;
}
  • Thread.h // 线程封装,内部有一个static变量记录当前创建了多少个线程
  • ThreadLocal.h // 线程局部变量封装,不用在使用pthread_get/setspecific
  • ThreadLocalSignleton.h // 线程局部单例,不用考虑多个线程同时创建
  • ThreadPool.h // 线程池包装,内部维护了一个异步队列,多个线程的工作就是取出task来执行
// 外部丢入Task放到线程池内部
void ThreadPool::run(const Task& task)
{
  if (threads_.empty()) // 如果没有任何线程
  {
    task();
  }
  else
  {
    MutexLockGuard lock(mutex_);
    queue_.push_back(task);
    cond_.notify();
  }
}
ThreadPool::Task ThreadPool::take()
{
  MutexLockGuard lock(mutex_);
  while (queue_.empty() && running_)
  {
    cond_.wait();
  }
  Task task;
  if(!queue_.empty())
  {
    task = queue_.front();
    queue_.pop_front();
  }
  return task;
}
// 线程回调函数
void ThreadPool::runInThread()
{
  try
  {
    while (running_)
    {
      Task task(take());
      if (task)
      {
        task();
      }
    }
  }
}

2 Buffer

Buffer设计的非常精巧。Buffer内部是一个std::vector<char*>表示的,逻辑上结构是这样的

/// @code
/// +-------------------+------------------+------------------+
/// | prependable bytes |  readable bytes  |  writable bytes  |
/// |                   |     (CONTENT)    |                  |
/// +-------------------+------------------+------------------+
/// |                   |                  |                  |
/// 0      <=      readerIndex   <=   writerIndex    <=     size
/// @endcode

头部有prependable bytes这个似乎可以不要,内部是占用8个字节,初始化的时候size占用了1024+8个字节。初始的时候readerIndex==writerIndex==8, 就好比现在没有任何数据写入。一旦开始要写入数据的话,那么writerIndex+=size(要写入的字节数)这个buffer会动态地增长。readerIndex标记的就是我们 可以读的下标,如果readerIndex==writerIndex就表示没有数据了。但是很明显这个Buffer并不是无限增长的,在makeSpace函数里面的话就可以看到,实际上 是会进行压缩的。

void makeSpace(size_t len)
{
    if (writableBytes() + prependableBytes() < len + kCheapPrepend) // 如果当前不能够通过压缩合并的话
    {
        buffer_.resize(writerIndex_+len);
    }
    else
    {
        // move readable data to the front, make space inside buffer
        assert(kCheapPrepend < readerIndex_); // 如果可以压缩的话那么就压缩
        size_t readable = readableBytes();
        std::copy(begin()+readerIndex_,
                  begin()+writerIndex_,
                  begin()+kCheapPrepend);
        readerIndex_ = kCheapPrepend;
        writerIndex_ = readerIndex_ + readable;
        assert(readable == readableBytes());
    }
}

3 Channel

class Channel : boost::noncopyable
{
  public:
    typedef boost::function<void()> EventCallback;
    typedef boost::function<void(Timestamp)> ReadEventCallback;
    void handleEvent(Timestamp receiveTime);
  private:
    EventLoop* loop_; // 属于哪一个Reactor
    const int  fd_; // 关联fd
    int        events_; // 关注事件
    int        revents_; // ready事件
    int        index_; // used by Poller. 在Poller中的编号,实际上没有特别意思

    boost::weak_ptr<void> tie_; // 绑定的对象,这个对于boost::weak_ptr不是很了解,但是这个对于理解框架没有用途
    bool tied_; // 是否绑定了对象上来
    bool eventHandling_; // 当前正在处理event
    ReadEventCallback readCallback_;
    EventCallback writeCallback_; // 定义如何写数据
    EventCallback closeCallback_; // 定义如何关闭连接
    EventCallback errorCallback_; // 定义如果出错的话如何处理
};

一旦EventLoop通知Channel触发事件的话那么就会调用handleEvent这个函数。参数receiveTime本身只对ReadEventCallback有效并且作为参数使用, 代表接收超时时间,对于write而言的话没有超时时间。内部的话handleEvent会根据revents触发的事件来分别决定调用哪些回调

void Channel::handleEventWithGuard(Timestamp receiveTime)
{
    eventHandling_ = true;
    if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
    {
        LOG_WARN << "Channel::handle_event() POLLHUP";
        if (closeCallback_) closeCallback_();
    }
    if (revents_ & POLLNVAL)
    {
        LOG_WARN << "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;
}

4 Poller

Poller本身也是一个抽象类,然后底层支持poll和epoll.

class Poller : boost::noncopyable
{
  public:
    typedef std::vector<Channel*> ChannelList;
    /// Polls the I/O events.
    /// Must be called in the loop thread.
    virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;  // 进行poll操作,活跃事件放在activeChannels里面
    /// Changes the interested I/O events.
    /// Must be called in the loop thread.
    virtual void updateChannel(Channel* channel) = 0; // 更新channel
    /// Remove the channel, when it destructs.
    /// Must be called in the loop thread.
    virtual void removeChannel(Channel* channel) = 0; // 删除channel
  private:
    EventLoop* ownerLoop_;
};

在poller目录下面有poll和epoll的对应实现,不过我们这里没有必要仔细阅读。需要注意的是这里的channel处理并没有引入优先级的概念。 poll操作的timeoutMs就是epoll_wait超时时间,而activeChannels就是活跃channel.返回值就是epoll_wait之后的时间戳。

5 EventLoop

和之前一样,我们还是看看EventLoop有哪些结构。对于EventLoop结构比较复杂,我们列出主要的接口和成员。 首先我们看EventLoop有runInLoop和queueInLoop功能,虽然作者建议event loop和一个线程绑定,但是在其他线程的话依然可以调用runInLoop和 queueInLoop的功能,将一些task加入到这个event loop对应的线程中执行。这样就很地然地引入了pendingFunctors字段。因为需要跨线程激活, 那么就需要线程之间的通知机制,这个使用eventfd来完成,对应字段就是wakeFd并且内部绑定了一个wakeupChannel.如果没有eventfd的话,通常也可以使用 pipe来完成。然后我们还允许向EventLoop里面添加定时器任务,就是runAt,runAfter和runEvery三个函数,我们只需要关注其中一个即可。

class EventLoop : boost::noncopyable
{
  public:
    typedef boost::function<void()> Functor;
    typedef boost::function<void()> TimerCallback;
    void loop();
    void quit();
    /// Runs callback immediately in the loop thread.
    /// It wakes up the loop, and run the cb.
    /// If in the same loop thread, cb is run within the function.
    /// Safe to call from other threads.
    void runInLoop(const Functor& cb);
    /// Queues callback in the loop thread.
    /// Runs after finish pooling.
    /// Safe to call from other threads.
    void queueInLoop(const Functor& cb);

    /// Runs callback at 'time'.
    /// Safe to call from other threads.
    ///
    TimerId runAt(const Timestamp& time, const TimerCallback& cb);
    ///
    /// Runs callback after @c delay seconds.
    /// Safe to call from other threads.
    ///
    TimerId runAfter(double delay, const TimerCallback& cb);
    ///
    /// Runs callback every @c interval seconds.
    /// Safe to call from other threads.
    ///
    TimerId runEvery(double interval, const TimerCallback& cb);
    ///
    /// Cancels the timer.
    /// Safe to call from other threads.
    ///
    // void cancel(TimerId timerId);
    // internal usage
    void wakeup();
    void updateChannel(Channel* channel);
    void removeChannel(Channel* channel);
    bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
  private:
    typedef std::vector<Channel*> ChannelList;
    bool looping_; /* atomic */
    bool quit_; /* atomic */
    bool eventHandling_; /* atomic */
    bool callingPendingFunctors_; /* atomic */
    const pid_t threadId_;
    Timestamp pollReturnTime_;
    boost::scoped_ptr<Poller> poller_;
    boost::scoped_ptr<TimerQueue> timerQueue_;
    int wakeupFd_;
    // unlike in TimerQueue, which is an internal class,
    // we don't expose Channel to client.
    boost::scoped_ptr<Channel> wakeupChannel_;
    ChannelList activeChannels_;
    MutexLock mutex_;
    std::vector<Functor> pendingFunctors_; // @BuardedBy mutex_
};

5.1 单线程单EventLoop

作者建议一个线程绑定一个EventLoop,这个实现呢?其实还是使用线程局部变量。首先定义线程局部变量

__thread EventLoop* t_loopInThisThread = 0;

然后在EventLoop构造函数的时候判断这个是否==0,如果不为=0的话说明在这个线程已经构造过一个EventLoop了。 直接使用__thread这个关键字,值得学习一下。

5.2 跨线程激活

首先我们看看跨线程激活是怎么操作的。在EventLoop的初始化函数内部初始化了wakeupFd并且创建了channel.但是如果不仔细阅读, 很可能觉得的这个channel没有注册。而实际上这个channel在enableReading()就会注册的。

EventLoop::EventLoop()
  : wakeupFd_(createEventfd()),
    wakeupChannel_(new Channel(this, wakeupFd_))
{
  wakeupChannel_->setReadCallback(
      boost::bind(&EventLoop::handleRead, this)); // 绑定到handleRead上面了
  // we are always reading the wakeupfd
  wakeupChannel_->enableReading();
}

一旦wakeup完成之后那么wakeUpFd_就是可读的,这样EventLoop就会被通知到并且立刻跳出epoll_wait开始处理。当然我们需要将这个wakeupFd_ 上面数据读出来,不然的话下一次又会被通知到,读取函数就是handleRead

void EventLoop::handleRead()
{
  uint64_t one = 1;
  ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
}

5.3 跨线程任务

runInLoop和queueInLoop就是跨线程任务。内容非常简单

void EventLoop::runInLoop(const Functor& cb){
  if (isInLoopThread()){ // 如果这个函数在自己的线程调用,那么就可以立即执行
    cb();
  }else{
    queueInLoop(cb); // 如果是其他线程调用,那么加入到pendingFunctors里面去
    wakeup(); // 并且通知这个线程,有任务到来
  }
}

void EventLoop::queueInLoop(const Functor& cb){
  {
  MutexLockGuard lock(mutex_);
  pendingFunctors_.push_back(cb);
  }
  if (isInLoopThread() && callingPendingFunctors_){
    wakeup(); // 被排上队之后如果是在自己线程并且正在执行pendingFunctors的话,那么就可以激活
    // 否则下一轮完全可以被排上,所以没有必要激活
  }
}

5.4 定时器任务

定时器任务都是交给了TimerQueue来处理的,在TimerQueue这个部分我们会简要地分析一下

TimerId EventLoop::runAt(const Timestamp& time, const TimerCallback& cb)
{
  return timerQueue_->addTimer(cb, time, 0.0); // time是在之后什么时候开始,0.0表示以后每次运行时间(0.0表示不会repeat).
}

5.5 How it works

基本上和hpserver非常相似,不断地调用poller::poll方法,然后在外层不断地查看是否需要quit.poll之后会得到activeChannels.和hpserver不同的是, muduo没有调用器(其实也不需要,本来就没有优先级概念),仅仅遍历这个activeChannels,并且调用内部的handleEvent方法,然后在调用pengdingFunctors 一些跨线程任务。

6 TimerQueue

TimerQueue里面最主要的方法就是addTimer.我们看看addTimer里面做了哪些事情,整个过程有点绕

TimerId TimerQueue::addTimer(const TimerCallback& cb,
                             Timestamp when,
                             double interval)
{
  Timer* timer = new Timer(cb, when, interval); // 首先创建一个Timer对象,然后将cb放在里面。内部有一个run函数,调用的就是cb
  loop_->runInLoop(
      boost::bind(&TimerQueue::scheduleInLoop, this, timer)); // 然后将这个timer丢到eventLoop里面去执行
  return TimerId(timer, timer->sequence());
}
void TimerQueue::scheduleInLoop(Timer* timer)
{
  loop_->assertInLoopThread();
  bool earliestChanged = insert(timer); // 将timer插入到内部的链表里面去,按照超时时间顺序插入,并且判断这个插入是否会影响最早时间

  if (earliestChanged)
  {
    resetTimerfd(timerfd_, timer->expiration()); // 如果影响的话,那么要修改这个timerfd超时时间。
  }
}

然后一旦timerfd可读的时候,就会调用下面这个函数

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(); // 对于这些超时的Timer,执行run()函数,对应也就是我们一开始注册的回调函数。
  }
  reset(expired, now);
}

7 EventLoopThread

EventLoopThread就是将一个EventLoop和Thread包装在一起的对象。这个内容到没有什么,不过觉得代码方面有点技巧。 我们在启动startLoop这个样就会执行线程threadFunc,但是我们必须等待threadFunc将栈上面的EventLoop绑定之后才可以返回,所以这里用到了条件变量。

EventLoop* EventLoopThread::startLoop(){
    thread_.start();
    {
        MutexLockGuard lock(mutex_);
        while (loop_ == NULL)
        {
            cond_.wait();
        }
    }
    return loop_;
}

void EventLoopThread::threadFunc(){
    EventLoop loop;
    {
        MutexLockGuard lock(mutex_);
        loop_ = &loop;
        cond_.notify();
    }
    loop.loop();
}

而EventLoopThreadPool就是维持一个EventLoopThread线程池,所以没有什么特别好说的。我们只需要setThreadNum告诉开多少个线程,然后调用start即可。

8 Acceptor

Acceptor帮助简化了搭建服务器accept这个部分的逻辑。通常这个逻辑是在单个线程里面完成的,所以抽取出来蛮有必要的。 代码不是很麻烦,用户要做的就是编写一个回调,这个回调在新建立连接时候出发,参数分别是链接fd和连接地址。

typedef boost::function<void (int sockfd,const InetAddress&)> NewConnectionCallback;

原理很简单,初始化socket和对应的channel并且监听READ事件,然后开始进行listen.一旦触发read事件的话那么就证明我们无阻塞 地进行accept,然后在READ事件回调里面进行accept。一旦accept成功的话就调用这个回调函数即可。

9 Connector

Connector也是为了简化客户端编写,用户只需要提供这个逻辑即可,这个回调函数在建立链接成功的时候使用

typedef boost::function<void (int sockfd)> NewConnectionCallback;

Connector初始化以EventLoop和服务器地址初始化,然后在start的时候的话开始尝试进行connect.如果返回非阻塞的错误码的话,那么 创建一个channel并且监视WRITE和ERROR事件,否则就会尝试重连(按照一定时间间隔).在Connector::handleWrite里面的话会将这个channel 移除,然后调用NewConnectionCallback来处理连接建立的事件。

10 TcpConnection

TcpConnection完成的工作就是当TCP连接建立之后处理socket的读写以及关闭。同样我们看看TcpConnection的结构

class TcpConnection : boost::noncopyable,
                      public boost::enable_shared_from_this<TcpConnection>
{
  public:
    /// Constructs a TcpConnection with a connected sockfd
    ///
    /// User should not create this object.
    TcpConnection(EventLoop* loop, // 建立连接需要一个Reactor
                  const string& name, // 连接名称
                  int sockfd, // 连接fd.
                  const InetAddress& localAddr, // 连接的address.
                  const InetAddress& peerAddr);
    // called when TcpServer accepts a new connection
    void connectEstablished();   // should be called only once
    // called when TcpServer has removed me from its map
    void connectDestroyed();  // should be called only once
  private:
    enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };
    void sendInLoop(const void* message, size_t len); // 发送消息
    void setState(StateE s) { state_ = s; }

    EventLoop* loop_;
    string name_;
    StateE state_;  // FIXME: use atomic variable
    // we don't expose those classes to client.
    boost::scoped_ptr<Socket> socket_; // socket.
    boost::scoped_ptr<Channel> channel_; // 连接channel
    InetAddress localAddr_;
    InetAddress peerAddr_;
    ConnectionCallback connectionCallback_; // 连接回调,这个触发包括在连接建立和断开都会触发
    MessageCallback messageCallback_; // 有数据可读的回调
    WriteCompleteCallback writeCompleteCallback_; // 写完毕的回调
    CloseCallback closeCallback_; // 连接关闭回调
    Buffer inputBuffer_; // 数据读取buffer.
    Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.
    boost::any context_; // 上下文环境
    // FIXME: creationTime_, lastReceiveTime_
    //        bytesReceived_, bytesSent_
};

首先TcpConnection在初始化的时候会建立好channel.然后一旦TcpClient或者是TcpServer建立连接之后的话,那么调用TcpConnection::connectEstablished. 这个函数内部的话就会将channel设置成为可读。一旦可读的话那么TcpConnection内部就会调用handleRead这个动作,内部托管了读取数据这个操作。 读取完毕之后然后交给MessageBack这个回调进行操作。如果需要写的话调用sendInLoop,那么会将message放在outputBuffer里面,并且设置可写。 后当可写的话TcpConnection内部就托管写,然后写完之后的话会发生writeCompleteCallback这个回调。托管的读写操作都是非阻塞的。如果希望断开的话调用 shutdown。解除这个连接的话那么可以调用TcpConnection::connectDestroyed,内部大致操作就是从reactor移除这个channel.

在TcpConnection这层并不知道一次需要读取多少个字节,这个是在上层进行消息拆分的,这点可以仔细阅读一下Httpserver这个example. TcpConnection一次最多读取64K字节的内容,然后交给上层。上层决定这些内容是否足够,如果不够的话那么直接返回让Reactor继续等待读。 同样写的话内部也是会分多次写。这样就要求reactor内部必须使用水平触发而不是边缘触发。

11 TcpClient

一旦我们了解了TcpConnection之后的话,这个托管了建立好连接之后所需要的处理的所有事情,那么我们对于client关心的重点就是如果触发连接的建立以及连接是如何断开的。

TcpClient::TcpClient(EventLoop* loop,
                     const InetAddress& serverAddr,
                     const string& name)
        : loop_(CHECK_NOTNULL(loop)),
          connector_(new Connector(loop, serverAddr)),
          name_(name),
          connectionCallback_(defaultConnectionCallback),
          messageCallback_(defaultMessageCallback),
          retry_(false),
          connect_(true),
          nextConnId_(1)
{
    connector_->setNewConnectionCallback(
        boost::bind(&TcpClient::newConnection, this, _1));
    // FIXME setConnectFailedCallback
}

可以看到初始化了connector这个对象并且设置了connector的连接建立的回调。我们需要设置一下TcpConnection所需要设置的回调之后,然后调用connect()这个方法。 内部会调用connector::start方法,一旦连接建立成功的话那么会调用TcpClient::newConnection这个函数。在这个函数内部会建立TcpConnection,并且调用 TcpConnection::connectEstablished,之后的所有操作都交给TcpConnection了。如果需要断开连接的话调用disconnect,内部会调用TcpConnection::shutdown.在析构 函数里面会调用TcpConneciton::connectDestroyed来移除连接。

12 TcpServer

从分析上我们和TcpClient一样只是关心连接是如何建立这个过程。

TcpServer::TcpServer(EventLoop* loop,
                     const InetAddress& listenAddr,
                     const string& nameArg)
  : loop_(CHECK_NOTNULL(loop)),
    hostport_(listenAddr.toHostPort()),
    name_(nameArg),
    acceptor_(new Acceptor(loop, listenAddr)),
    threadPool_(new EventLoopThreadPool(loop)),
    connectionCallback_(defaultConnectionCallback),
    messageCallback_(defaultMessageCallback),
    started_(false),
    nextConnId_(1)
{
  acceptor_->setNewConnectionCallback(
      boost::bind(&TcpServer::newConnection, this, _1, _2));
}

同样是建立好acceptor这个对象然后设置好回调为TcpServer::newConnection,同时在外部设置好TcpConnection的各个回调。然后调用start来启动服务器,start 会调用acceptor::listen这个方法,一旦有连接建立的话那么会调用newConnection.下面是newConnection代码

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
    loop_->assertInLoopThread();
    EventLoop* ioLoop = threadPool_->getNextLoop();
    char buf[32];
    snprintf(buf, sizeof buf, ":%s#%d", hostport_.c_str(), nextConnId_);
    ++nextConnId_;
    string connName = name_ + buf;
    // FIXME poll with zero timeout to double confirm the new connection
    TcpConnectionPtr conn(
        new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));
    connections_[connName] = conn;
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);
    conn->setCloseCallback(
        boost::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
    ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
}

对于服务端来说连接都被唯一化了然后映射称为字符串放在connections_这个容器内部。threadPool_->getNextLoop()可以轮询地将取出么一个线程然后将 TcpConnection::connectEstablished轮询地丢到每个线程里面去完成。存放在connections_是有原因了,每个TcpConnection有唯一一个名字,这样Server 就可以根据TcpConnection来从自己内部移除链接了。在析构函数里面可以遍历connections_内容得到所有建立的连接并且逐一释放。

comments powered by Disqus