TcpServer::start都做了些什么

在使用muduo库的时候，在TcpServer.start()后执行SetThreadNum()是没有作用的。
因为在TcpServer.start()后线程数量就已经定下了，这时后去修改线程数量是没有意义的。

start()是muduo对整个Multiple Reactors模型的初始化，主要做的就是accept前的准备，执行流程如下：

1. 首先就是根据SetThreadNum设置的线程数量，创建对应数量Thread
2. 启动所有Thread，在一个Thread中开启一个EventLoop并进入循环（one loop per thread就体现在这）
3. accept前的准备，mainLoop进入listen状态

EventLoopThreadPool

muduo的Multiple Reactors模型是epoll+线程池的方式，EventLoopThreadPool的作用就是初始化Multiple Reactors，并提供connfd的分配策略。
他需要包含mainReactor的EventLoop（在设置单Reactor时会用到），EventLoop和Thread的映射（one loop per thread），以及所有的subReactor的EventLoop（connfd分配策略会用到）。
重写EventLoopThreadPool.h：

#pragma once
#include "noncopyable.h"

#include 
#include 
#include 
#include 

class EventLoop;
class EventLoopThread;

class EventLoopThreadPool : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop*)>;

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

    void setThreadNum(int numThreads) { numThreads_ = numThreads; }

    void start(const ThreadInitCallback& cb = ThreadInitCallback());

    //如果工作在多线程中，baseLoop_默认以轮询的方式分配channel给subloop
    EventLoop* getNextLoop();

    std::vector<EventLoop*> getAllLoop();

    bool started() const { return started_; }

    const std::string name() const { return name_; }
private:
    EventLoop* baseLoop_; //eventloop loop
    std::string name_;
    bool started_;
    int numThreads_;
    int next_;
    std::vector<std::unique_ptr<EventLoopThread>> threads_;
    std::vector<EventLoop*> loops_;
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

重写EventLoopThreadPool.cc：

#include "EventLoopThreadPool.h"
#include "EventLoopThread.h"

#include 

EventLoopThreadPool::EventLoopThreadPool(EventLoop* baseLoop, const std::string& nameArg)
    :baseLoop_(baseLoop)
    ,name_(nameArg)
    ,started_(false)
    ,numThreads_(0)
    ,next_(0)
{}

EventLoopThreadPool::~EventLoopThreadPool()
{}

void EventLoopThreadPool::start(const ThreadInitCallback& cb)
{
    started_ = true;   

    for(int i = 0; i < numThreads_; i++)
    {
        char buf[name_.size() + 32]; 
        snprintf(buf, sizeof buf, "%s%d", name_.c_str(), i);
        EventLoopThread *t = new EventLoopThread(cb, buf);
        threads_.push_back(std::unique_ptr<EventLoopThread>(t));
        loops_.push_back(t->startLoop());   //底层创建线程，绑定一个新的EventLoop，并返回该loop的地址
    }

    if(numThreads_ == 0 && cb)
    {
        cb(baseLoop_);
    }
}

//如果工作在多线程中，baseLoop_默认以轮询的方式分配channel给subloop
EventLoop* EventLoopThreadPool::getNextLoop()
{
    EventLoop* loop = baseLoop_;

    if(!loops_.empty())
    {
        loop = loops_[next_];
        ++next_;
        if(next_ >= loops_.size())
        {
            next_ = 0;
        }
    }

    return loop;
}

std::vector<EventLoop*> EventLoopThreadPool::getAllLoop()
{
    if(loops_.empty())
    {
        return std::vector<EventLoop*>(1,baseLoop_);
    }
    else
    {
        return loops_;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

EventLoopThread

EventLoopThread是one loop per thread 的体现，维护了Thread和EventLoop的映射关系。
需要注意，创建EventLoop和记录EventLoop是在不同线程间执行的，需要保证其执行的线程同步互斥。
重写EventLoopThread.h：

#pragma once
#include "noncopyable.h"
#include "Thread.h"

#include 
#include 
#include 
#include 

class EventLoop;

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

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

    EventLoop* startLoop();
private:
    void threadFunc();

    EventLoop* loop_;
    bool exiting_;
    Thread thread_;
    std::mutex mutex_;
    std::condition_variable cond_;
    ThreadInitCallback callback_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

重写EventLoopThread.cc：

#include "EventLoopThread.h"
#include "EventLoop.h"

EventLoopThread::EventLoopThread(const ThreadInitCallback& cb, const std::string& name)
    :loop_(nullptr)
    ,exiting_(false)
    ,thread_(std::bind(&EventLoopThread::threadFunc, this), name)
    ,mutex_()
    ,cond_()
    ,callback_(cb)
{}

EventLoopThread::~EventLoopThread()
{
    exiting_ = true;
    if(loop_ != nullptr)
    {
        loop_->quit();
        thread_.join();
    }
}

EventLoop* EventLoopThread::startLoop()
{
    thread_.start(); //启动底层的新线程

    EventLoop *loop = nullptr;
    {
        std::unique_lock<std::mutex> lock(mutex_);
        while (loop_ == nullptr)
        {
            cond_.wait(lock);
        }
        loop = loop_;
    }
    return loop;
}

//下面这个方法，是在单独的新线程里面运行的
void EventLoopThread::threadFunc()
{
    EventLoop loop; //创建一个独立的eventLoop，和上面的线程是一一对应的，one loop per pthread

    if(callback_)
    {
        callback_(&loop);
    }
    
    {
        std::unique_lock<std::mutex> lock(mutex_);
        loop_ = &loop;
        cond_.notify_one();
    }
    loop.loop(); //EventLoop loop => Poller.poll
    loop_ = nullptr;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

Thread

EventLoop执行事件循环需要一个线程，而Thread类用于提供线程。
重写Thread.h：

#pragma once

#include "noncopyable.h"

#include 
#include 
#include 
#include 
#include 
#include 

class Thread : noncopyable
{
public:
    using ThreadFunc = std::function<void()>;

    explicit Thread(ThreadFunc, const std::string &name = std::string());

    ~Thread();

    void start();
    void join();

    bool started()const { return started_; }
    pid_t tid() const { return tid_; }
    const std::string& name() const { return name_; }

    static int numCreated() { return numCreated_; }
private:
    void setDefaultName();

    bool started_;
    bool joined_;
    std::shared_ptr<std::thread> thread_;
    pid_t tid_;
    ThreadFunc func_;
    std::string name_;
    static std::atomic_int numCreated_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

重写Thread.cc：

#include "Thread.h"
#include "CurrentThread.h"

#include 

std::atomic_int Thread::numCreated_(0);

Thread::Thread(ThreadFunc func, const std::string &name)
    :started_(false)
    ,joined_(false)
    ,tid_(0)
    ,func_(std::move(func))
    ,name_(name)
{
    setDefaultName();
}

Thread::~Thread()
{
    if(started_ && !joined_)
    {
        thread_->detach(); //thread类提供的设置分离线程的方法
    }
}

void Thread::start()    //一个Thread对，记录的就是一个新线程的详细信息
{
    sem_t sem;
    sem_init(&sem, false, 0);
    started_ = true;
    //开启线程
    thread_ = std::shared_ptr<std::thread>(new std::thread([&](){ 
        //获取线程的tid值
        tid_ = CurrentThread::tid();  
        sem_post(&sem);
        func_();    //开启一个新线程，专门执行该线程函数
    }));

    //等待获取上面新创建的线程的tid值
    sem_wait(&sem);
}
void Thread::join()
{
    joined_ = true;
    thread_->join();
}
void Thread::setDefaultName()
{
    int num = ++numCreated_;
    if(name_.empty())
    {
        char buf[32] = {0};
        snprintf(buf,sizeof buf, "Thread%d", num);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55