EventLoop的多线程应用

前面讲的EventLoop为一个IO线程提供运行循环（loop），那muduo库如何支持多线程呢？

• EventLoopThread IO线程类
• EventLoopThreadPool IO线程池类

IO线程池的功能是开启若干个IO线程，并让这些IO线程处于线程循环的状态。

也就是说，EventLoop实现one loop per thread模型中的loop，EventLoopThread 实现的是per thread，EventLoopThreadPool 实现的是multi-thread环境下：one loop per thread。

多个Reactor模型

图中每个Reactor都是一个线程，mainReactor通常是main线程，关注监听套接字；subReactor关注的是连接套接字。如果没有subReactor，所有跟监听套接字、连接套接字有关的事件，都交由mainReactor处理。

为了简便，本文下面所有EventLoopThreadPool （事件循环线程池） 统一简称IO线程池或线程池，EventLoopThread 统一简称IO线程。

EventLoopThreadPool 事件循环线程池类

EventLoopThreadPool 事件循环线程池类对象通常由main线程创建，绑定main线程创建的EventLoop（即baseLoop_），对应mainReactor。该线程池根据用户指定线程数，创建EventLoopThread对应subReactor。
注意：EventLoopThreadPool 属于Reactor的一部分，但不等于某个Reactor。

EventLoopThreadPool类声明

class EventLoopThreadPool : noncopyable // 作为线程池对象，绑定了背后的系统资源，如线程，因此是引用语义（不可拷贝）
{
public:
    typedef std::function<void(EventLoop*)> ThreadInitCallback;

    EventLoopThreadPool(EventLoop* baseLoop, const std::string& nameArg);
    ~EventLoopThreadPool();

    /* 设置线程数量, 需要在start()之前调用 */
    void setThreadNum(int numThreads)
    { numThreads_ = numThreads; }

    /* 启动线程池, 设置线程函数初始回调 */
    void start(const ThreadInitCallback& cb = ThreadInitCallback());

    /*
     * valid after calling start()
     * round-robin(轮询)
     */
    EventLoop* getNextLoop();

    /*
     * With the same hash code, it will always return the same EventLoop.
     */
    EventLoop* getLoopForHash(size_t hashCode);

    /* 获取所有loops(EventLoop数组) */
    std::vector<EventLoop*> getAllLoops();

    /* 获取线程池启动状态 */
    bool started() const
    { return started_; }

    /* 获取线程池名称 */
    const std::string& name() const
    { return name_; }

private:

    EventLoop* baseLoop_; // 与Acceptor所属EventLoop相同
    std::string name_;    // 线程池名称, 通常由用户指定. 线程池中EventLoopThread名称依赖于线程池名称
    bool started_;   // 线程池是否启动标志
    int numThreads_; // 线程数
    int next_;       // 新连接到来，所选择的EventLoopThread下标
    std::vector<std::unique_ptr<EventLoopThread>> threads_; // IO线程列表
    std::vector<EventLoop*> loops_;  // EventLoop列表, 指向的是EventLoopThread线程函数创建的EventLoop对象
};

EventLoopThreadPool的构造与析构

构造函数很简单，对需baseLoop_、name_、started等进行了初始化。

有个问题一直困扰自己：为什么EventLoop指针 baseLoop，没有通过智能指针管理内存？
这里可以得到解决，因为baseLoop通常是main线程创建的栈变量，loops_数组（std::vector<EventLoop*>）中的EventLoop对象是线程函数创建的栈变量。当离开线程作用域时，栈变量会自动释放。因此，在这之前，不要delete loop对象。

EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseLoop, const std::string &nameArg)
: baseLoop_(baseLoop), // 指向基础的EventLoop对象
name_(nameArg),        // 线程池名称
started_(false),       // 启动状态
numThreads_(0),        // 线程数量
next_(0)               // 下一个EventLoopThread位于threads_数组中的下标
{
}

EventLoopThreadPool::~EventLoopThreadPool()
{
  // Don't delete loop, it's stack variable
}

start() 启动IO线程池

IO线程池在创建后，通过调用start()启动线程池。主要工作：
1）确保baseLoop所属线程调用start；
2）创建用户指定线程组，启动线程组线程，并记录子线程对应EventLoop；
3）如果没有指定线程数量（或为指定0），调用用户指定的线程函数初始回调。

/**
* 启动IO线程池.
* 只能启动一次, 而且必须是baseLoop_的创建线程调用start().
* @param cb 线程函数初始回调
*/
void EventLoopThreadPool::start(const ThreadInitCallback &cb)
{
    assert(!started_); // 防止重复启动线程池
    baseLoop_->assertInLoopThread(); // 断言baseLoop_对象创建者是线程池的start()调用者

    started_ = true; // 标记线程池已启动

    // 根据用户指定线程数, 创建IO线程组
    /* create numThreads_ EventLoopThread, added to threads_ */
    for (int i = 0; i < numThreads_; ++i) { // 线程编号范围取决于用户指定的线程数
        char buf[name_.size() + 32];
        snprintf(buf, sizeof(buf), "%s%d", name_.c_str(), i); // IO线程名称: 线程池名称 + 线程编号
        EventLoopThread* t = new EventLoopThread(cb, buf);
        threads_.push_back(std::unique_ptr<EventLoopThread>(t)); // 将EventLoopThread对象指针 插入threads_数组
        loops_.push_back(t->startLoop()); // 启动IO线程, 并将线程函数创建的EventLoop对象地址 插入loops_数组
    }
    if (numThreads_ == 0 && cb)
    { // 如果没有创建任何线程, 也会调用回调cb; 否则, 会在新建的线程函数初始化完成后(进入loop循环前)调用
        cb(baseLoop_);
    }
}

用户端TcpServer调用start()。可以看到threadInitCallback_是由TcpServer传入的，而TcpServer::threadInitCallback_是由更上一层级的用户传入。

/**
* 启动TcpServer, 初始化线程池, 连接接受器Accept开始监听(Tcp连接请求)
*/
void TcpServer::start()
{
    if (started_.getAndSet(1) == 0) // 防止多次重复启动
    {
        threadPool_->start(threadInitCallback_); // 启动线程池, 并设置线程初始化完成的回调函数

        assert(!acceptor_->listening());
        loop_->runInLoop(
                std::bind(&Acceptor::listen, get_pointer(acceptor_)));
    }
}

分派任务给IO线程的利器：getNextLoop()

每当有一个新Tcp连接建立时，TcpServer调用newConnection新建一个TcpConnection对象负责该连接。然而，如何将TcpConnection对象分派给一个IO线程对应的EventLoop对象呢？
这就可以利用getNextLoop()，从IO线程池维护的EventLoop数组loops_中轮询取得一个EventLoop对象，每次调用数组下标+1，这样得到负载均衡的目的。

/**
* 从线程池获取下一个event loop
* @note 默认event loop是baseLoop_ (创建baseLoop_线程, 通常也是创建线程池的线程).
* 没有调用setThreadNum()设置numThreads_(number of threads)时, numThreads_默认为０,
* 所有IO操作都默认交由baseLoop_的event loop来完成, 因为没有其他IO线程.
*/
EventLoop* EventLoopThreadPool::getNextLoop()
{
    baseLoop_->assertInLoopThread();
    assert(started_);
    EventLoop* loop = baseLoop_;

    // 如果loops_为空, 则loop指向baseLoop
    // 如果非空, 则按round-robin(RR, 轮叫)的调度方式(从loops_列表中)选择一个EventLoop
    if (!loops_.empty())
    {
        // round-robin
        loop = loops_[next_];
        ++next_;
        if (implicit_cast<size_t>(next_) >= loops_.size())
        {
            next_ = 0;
        }
    }
    return loop;
}

当然，TcpServer所谓分派TcpConnection给IO线程，实际上就是将TcpConnection对象在构造时，绑定到指定EventLoop对象。
部分代码见：

void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
{
    ...
    /* 从EventLoop线程池中，取出一个EventLoop对象构造TcpConnection对象，便于均衡各EventLoop负责的连接数　*/
    EventLoop* ioLoop = threadPool_->getNextLoop(); // next event loop from the event loop thread pool
    ...
    // FIXME poll with zero timeout to double confirm the new connection
    // FIXME use make_shared if necessary
    /* 新建TcpConnection对象,　并加入ConnectionMap */
    TcpConnectionPtr conn(new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));
    connections_[connName] = conn;
    ...
}

思考：getNextLoop() 为什么可以返回EventLoop 原生指针？*
答案同构造函数EventLoop没有通过智能指针管理，而是栈变量，这些栈变量只有在程序退出时，才会释放。因此，可以认为其生命周期是整个应用程序。

不常用的getLoopForHash，getAllLoops

getLoopForHash，getAllLoops这是两个备用接口，分别用于通过hashCode获取loops_数组中的EventLoop对象，获取整个EventLoop对象数组loops_。由于muduo库没有任何地方用到，这里不详述。

测试EventLoopThreadPool

思路：
3个测试用例：为IO线程池创建0个线程，1个线程，3个线程。然后，start线程池、并回调init，3个测试用例分别用getNextLoop调用多次，判断获得的EventLoop对象地址是否为期望值。

// EventLoopThreadPool_unittest.cpp
void print(EventLoop* p = NULL)
{
    printf("main(): pid=%d, tid=%d, loop=%pn",
          getpid(), CurrentThread::tid(), p);
}

void init(EventLoop* p)
{
    printf("init(): pid=%d, tid=%d, loop=%pn",
           getpid(), CurrentThread::tid(), p);
}

int main()
{
    print();

    EventLoop loop;
    loop.runAfter(11, std::bind(&EventLoop::quit, &loop));

    { // 0线程的IO线程池, 默认用main线程作为baseLoop所属线程, 处理IO事件
        printf("Single thread %p:n", &loop);
        EventLoopThreadPool model(&loop, "single"); // 0线程IO线程池
        model.setThreadNum(0);
        model.start(init); // 启动线程池并回调init
        // 从线程池连续取３次 EventLoop
        assert(model.getNextLoop() == &loop);
        assert(model.getNextLoop() == &loop);
        assert(model.getNextLoop() == &loop);
    }

    { // 单1线程的IO线程池
        printf("Another thread:n");
        EventLoopThreadPool model(&loop, "another"); // 1个线程的IO线程池
        model.setThreadNum(1);
        model.start(init);
        EventLoop* nextLoop = model.getNextLoop();
        nextLoop->runAfter(2, std::bind(print, nextLoop));
        assert(nextLoop != &loop);
        assert(nextLoop == model.getNextLoop());
        assert(nextLoop == model.getNextLoop());
        ::sleep(3);
    }

    { // 3线程的IO线程池
        printf("Three thread:n");
        EventLoopThreadPool model(&loop, "three");
        model.setThreadNum(3);
        model.start(init);
        EventLoop* nextLoop = model.getNextLoop();
        nextLoop->runInLoop(std::bind(print, nextLoop));
        assert(nextLoop != &loop);
        assert(nextLoop != model.getNextLoop());
        assert(nextLoop != model.getNextLoop());
        assert(nextLoop == model.getNextLoop()); // 3次以后循环回来
    }
    loop.loop();
    return 0;
}

EventLoopThread 事件循环线程类

一个EventLoopThread对象对应一个IO线程，而IO线程函数负责创建局部EventLoop对象，并启动EventLoop的loop循环。

class EventLoopThread : noncopyable
{
public:
    typedef std::function<void (EventLoop*)> ThreadInitCallback;

    EventLoopThread(const ThreadInitCallback& cb = ThreadInitCallback(),
                    const string& name = string());
    ~EventLoopThread();

    /* 启动IO线程函数中的loop循环, 返回IO线程中创建的EventLoop对象地址(栈空间) */
    EventLoop* startLoop();

private:
    void threadFunc(); // IO线程函数

    // 思考:这里为什么需要mutex_保护, 而EventLoopThreadPool::baseLoop_却不需要?
    EventLoop* loop_ GUARDED_BY(mutex_); // 绑定的EventLoop对象指针
    bool exiting_;  // 暂无特殊用途
    Thread thread_; // 线程, 用于实现IO线程中的线程功能
    MutexLock mutex_; // 互斥锁
    Condition cond_ GUARDED_BY(mutex_); // 条件变量
    ThreadInitCallback callback_;       // 线程函数初始回调
};

EventLoopThread的构造

EventLoopThread的结构很简单，对外只提供启动IO线程的接口startLoop()。
思考：为什么EventLoopThread的EventLoop对象指针loop_，需要互斥锁保护，而其他类如EventLoopThreadPool的EventLoop对象指针baseLoop_ 却不需要互斥锁保护？
因为EventLoopThread中的loop_指针在创建时（startLoop()中），会存在调用线程和子线程函数threadFunc同时读、写loop_的情况，也就是并发访问，因而需要互斥锁保护。
反观EventLoopThreadPool::baseLoop_，在构造后，就只是读操作，没有写baseLoop_指针本身。而且baseLoop_在构造时，也是调用者传入，不存在并发读写访问的问题。

EventLoopThread::EventLoopThread(const EventLoopThread::ThreadInitCallback &cb, const string &name)
: loop_(NULL),
  exiting_(false),
  thread_(std::bind(&EventLoopThread::threadFunc, this), name), // 注意这里只是注册线程函数, 名称, 并未启动线程函数
  mutex_(),
  cond_(mutex_),
  callback_(cb)
{
}

EventLoopThread::~EventLoopThread()
{
    exiting_ = true;
    // 不是100%没有冲突, 比如threadFunc中正运行callback_回调, 然后立即析构当前对象.
    // 此时, IO线程函数已经启动, 创建EventLoop了对象, 但还没有修改loop_, 此时loop_一直为NULL
    // 也就是说, 无法通过析构让IO线程退出loop循环, 也无法连接线程.
    if (loop_ != NULL) // not 100% race-free, eg. threadFunc could be running callback_
    {
        // sitll a tiny change to call destructed object, if threadFunc exists just now.
        // but when EventLoopThread destructs, usually programming is exiting anyway.
        loop_->quit(); // 退出IO线程loop循环
        thread_.join(); // 连接线程, 回收资源
    }
}

startLoop 启动IO线程

启动IO线程的过程很简单，就是启动一个IO线程，然后（调用线程）等待IO线程函数初始化完成。

/**
* 启动IO线程(函数), 运行EventLoop循环
* @return 返回EventLoop*, 实际上是线程函数threadFunc创建的EventLoop类型局部变量
*/
EventLoop* EventLoopThread::startLoop()
{
    assert(!thread_.started()); // avoid repeated start loop
    thread_.start(); // 启动线程

    EventLoop* loop = NULL;
    {
        MutexLockGuard lock(mutex_);
        while (loop_ == NULL)
        {
            cond_.wait();  // 同步等待线程函数完成初始化工作, 唤醒等待在此处的调用线程
        }
        loop = loop_;
    }

    return loop;
}

IO线程函数threadFunc

IO线程函数主要工作：创建EventLoop局部对象, 运行loop循环。

思考：最后为什么要清除loop_?
因为loop_可能在其他地方访问，而IO线程函数退出时，线程已经不能继续运行，代表IO线程EventLoop的loop_也就没有了存在意义。
如果不清空，析构函数可能会导致重复调用loop_->quit()，让IO线程loop循环重复退出。

这里带来一个新问题，可能是muduo库的一个bug：
如果在其他地方通过loop_->quit()让IO线程退出loop循环，而loop_置为NULL，那么线程资源在哪通过join/detach，以回收线程资源？
显然不是析构函数，因为析构函数中IO线程join的前提是loop非NULL。

/**
* IO线程函数, 创建EventLoop局部对象, 运行loop循环
*/
void EventLoopThread::threadFunc()
{
    EventLoop loop; // 创建线程函数局部EventLoop对象, 只有线程函数退出, EventLoop::loop()退出时, 才会释放该对象

    if (callback_) // 运行线程函数初始回调
    {
        callback_(&loop);
    }

    {
        MutexLockGuard lock(mutex_);
        loop_ = &loop;
        cond_.notify(); // 唤醒等待在cond_条件上的线程(i.e. startLoop的调用线程)
    }

    loop.loop(); // 运行IO线程循环, 即事件循环, 通常不会退出, 除非调用EventLoop::quit
    // assert(exiting_);
    MutexLockGuard lock(mutex_);
    loop_ = NULL; // 思考: 最后为什么要清除loop_?
}

测试EventLoopThread

思路：
3个测试用例：
1）不启动的EventLoopThread对象；
2）利用析构调用quit()，退出IO线程的loop循环；
3）在析构前，调用quit()退出IO线程的loop循环；

// EventLoopThread_unittes.cc
void print(EventLoop* p = NULL)
{
    printf("print: pid=%d, tid=%d, loop=%pn",
           getpid(), CurrentThread::tid(), p);
}

void quit(EventLoop* p)
{
    print(p);
    p->quit();
}

int main()
{
    print();

    {
        EventLoopThread thr1; // never start
    }

    {
        // dtor calls quit()
        EventLoopThread thr2;
        EventLoop* loop = thr2.startLoop();
        loop->runInLoop(std::bind(print, loop));
        CurrentThread::sleepUsec(500 * 1000);
    }

    {
        // quit() before dtor
        EventLoopThread thr3;
        EventLoop* loop = thr3.startLoop();
        loop->runInLoop(std::bind(quit, loop));
        CurrentThread::sleepUsec(500 * 1000);
    }
    return 0;
}

知识点

互斥锁 + 条件变量等待指定条件

一个线程通过while语句 + cond_.wait() 的形式，等待指定条件（loop_ != NULL），防止虚假唤醒。另外一个线程在条件满足后，通过cond_唤醒等待线程。
这样，可以有效达到一个线程等待另一个线程的目的。当然，这里也可以用门栓CountDownLatch，这里用互斥锁+条件变量也可以实现同样目的。

// 线程1
MutexLockGuard lock(mutex_);
while (loop_ == NULL)
{
    cond_.wait();  // 等待线程函数完成初始化工作, 唤醒等待在此处的调用线程
}

// 线程2
EventLoop loop;
MutexLockGuard lock(mutex_);
loop_ = &loop;
cond_.notify(); // 唤醒等待在cond_条件上的线程(i.e. startLoop的调用线程)