muduo库剖析(1)

muduo库剖析(1)

muduo网络库

项目地址：https://github.com/huerni/cgmuduo.git

muduo是一个基于非阻塞IO和事件驱动的C++网络库，采用Reactor模式和one loop per thread + thread pool线程模型。

类功能介绍

Channel.*，Poller.*，EventLoop.* 三类为一体，在底层负责事件循环。EventLoop包含Channel，Poller，EventLoop负责轮询访问Poller，得到激活Channel列表，使Channel自己根据自身情况调用相应回调。

Thread.*，EventLoopThread.*，EventLoopThreadPool.*将线程与EventLoop绑定在一起，实现one loop per thread + thread pool线程模型。

TcpServer.*，Acceptor.*，TcpConnection.*为上层Reactor模式的实现。TcpServer面向用户，可由用户自定义回调方法以及启动服务器。Acceptor也看做分发器，当有新连接到来时，用来将连接转发给子线程处理。TcpConnection在子线程处理被分来的连接的具体事件。

执行流程

1. 用户创建TcpServer对象调用start()开启服务器，启动线程池，启动一个baseLoop(主线程)，由runInLoop()处理Acceptor(Reactor)的listen()方法，进行监听操作；
2. 一旦有客户端请求连接，Accetor调用TcpServer设置的newConnectionCallback_，将请求分发给subLoop(子线程)，即从线程池中取出，然后将连接丢给子线程，由TcpConnection处理，此后该连接的所有操作都由该子线程独自处理，与其他线程无关。
3. TcpConnection将读，写，关闭，错误操作与回调填充至绑定的Channel，然后调用runInLoop()和queueInLoop()由底层事件循环处理。
4. 底层事件循环中，主线程称为baseLoop，只执行监听操作与用户回调。而子线程具体执行操作由Channel具体返回的事件所执行。由图所示，蓝色部分由三类组成，EventLoop，Channel和Poller。每个EventLoop分别与各自的线程绑定，同时将Poller和Channel连接起来，调用loop()方法将Channel的结果送入Poller执行，将Poller返回的结果各自送入Channel执行。

TcpServer类

TcpServer类面向用户，用户在编码服务器时，对TcpServer类进行创建和设置。

TcpServer类包含如下数据成员：

private: 
	using ConnectionMap = std::unordered_map;
	// 首先TcpServer含有一个baseLoop，此Loop绑定主线程，只负责监听操作。
    EventLoop *loop_; // baseloop
    
	// 服务器信息
    const std::string ipPort_;
    const std::string name_;
	
	// 含有acceptor，类似于分发起，封装监听操作
    std::unique_ptr acceptor_;
    
	// 服务器含有的线程池
    std::shared_ptr threadPool_;
	
	// 连接回调，读写回调，由用户编写
    ConnectionCallback connectionCallback_;
    MessageCallback messageCallback_;
    WriteCompleteCallback writeCompleteCallback_;
	
	// 线程初始化时调用回调，可选
    ThreadInitCallback threadInitCallback_;
	
	// 标志是否启动
    std::atomic_int started_;
	
	// 保存所有连接
    int nextConnId_;
    ConnectionMap connections_;
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

用户创建TcpServer对象时，需指定baseLoop，需要监听的套接字，并设置自定义的回调和线程池中需要初始化的线程个数。

TcpServer::TcpServer(EventLoop* loop, 
                    const InetAddress &listenAddr, 
                    const std::string &nameArg, 
                    Option option)
                : loop_(CheckLoopNotNull(loop))
                , ipPort_(listenAddr.toIpPort())
                , name_(nameArg)
                , acceptor_(new Acceptor(loop, listenAddr, option == kReusePort))
                , threadPool_(new EventLoopThreadPool(loop, name_))
                , connectionCallback_()
                , messageCallback_()
                , nextConnId_(1)
                , started_(0) 
{
    // 传入新连接到来操作回调到acceptor
    acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection, this, 
        std::placeholders::_1, std::placeholders::_2));
}

void setThreadInitcallback(const ThreadInitCallback &cb) { threadInitCallback_ = cb; }
void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; }
void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; }
void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; }

void setThreadNum(int numThreads);
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

设置好后，调用start()开启服务器

void TcpServer::start() {
    if(started_++ == 0) {
        // 开启线程池
        threadPool_->start(threadInitCallback_);
        // 由loop轮询运行acceptor封装的listen函数
        loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
    }
}
1
2
3
4
5
6
7
8

当新连接到来时，会调用TcpServer中的newConnection方法(怎么调用？可查看Acceptor.*代码)

void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr) {
    // 从线程池中获取空闲线程
    EventLoop *ioLoop = threadPool_->getNextLoop();
    char buf[64] = {0};
    snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
    ++nextConnId_;
    std::string connName = name_ + buf;

    LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s \n", 
    name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str());

    sockaddr_in local;
    ::bzero(&local, sizeof local);
    socklen_t addrlen = sizeof local;
    if(::getsockname(sockfd, (sockaddr*)&local, &addrlen) < 0) {
        LOG_ERROR("sockets::getLocalAddr");
    }
    InetAddress localAddr(local);
	// 创建TcpConnection对象，并将自定义回调传入
    TcpConnectionPtr conn(new TcpConnection(
                            ioLoop,
                            connName,
                            sockfd,
                            localAddr,
                            peerAddr));

    connections_[connName] = conn;
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);

    conn->setCloseCallback(
        std::bind(&TcpServer::removeConnection, this, std::placeholders::_1)
    );
	// 丢入loop中轮询连接建立完成回调
    ioLoop->runInLoop(std::bind(&TcpConnection::connectEstableished, conn));
}
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

Acceptor类

muduo采用Reactor模式，主线程只负责监听连接，新的连接丢给子线程操作。而Acceptor就是封装了监听连接操作的类，且该类只在主线程执行。

TcpServer创建时，便会创建所拥有的Acceptor类，并设置新连接带来的回调。

Acceptor类所拥有的数据成员：

private:
	EventLoop *loop_;  // 所绑定的主线程
	Socket acceptSocket_;  // 负责监听的socket
	Channel acceptChannel_;  // 负责监听的channel
	newConnectionCallback newConnectionCallback_;  // 新连接到来的回调
	bool listenning_;  // 是否在监听
1
2
3
4
5
6

TcpServer将需要监听的地址和接口传入acceptor

// 创建负责监听socket
static int createNonblocking() {
    int sockfd = ::socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if(sockfd < 0) {
        LOG_FATAL("%s:%s:%d listen socket create err:%d \n", __FILE__, __FUNCTION__, __LINE__, errno);
    }
    
    return sockfd;
}

Acceptor::Acceptor(EventLoop *loop, const InetAddress &listenAddr, bool reuseport) 
        : loop_(loop)
        , acceptSocket_(createNonblocking())
        , acceptChannel_(loop_, acceptSocket_.fd()), listenning_(false) {
    acceptSocket_.setReuseAddr(true);
    acceptSocket_.setReusePort(true);
    acceptSocket_.bindAddress(listenAddr); // 绑定地址
    // 当有新用户到来时，负责监听的channel所需要执行的回调，即handleRead函数
    acceptChannel_.setReadCallback(std::bind(&Acceptor::handleRead, this));  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

TcpServer在其构造函数中设置newConnection回调，当新连接到来时，便会执行TcpServer中的newConnection方法

// 有连接事件到来
void Acceptor::handleRead(){
    InetAddress peerAddr;
    int connfd = acceptSocket_.accept(&peerAddr);
    if(connfd >= 0) {
        // TcpServer传入的回调
        if(newConnectionCallback_) {
            newConnectionCallback_(connfd, peerAddr);
        }
        else {
            ::close(connfd);
        }
    }
    else {
        LOG_FATAL("%s:%s:%d accept err:%d \n \n", __FILE__, __FUNCTION__, __LINE__, errno);
        if(errno == EMFILE) { // too many open files
            LOG_ERROR("%s:%s:%d sockfd reached limit! \n", __FILE__, __FUNCTION__, __LINE__);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

TcpConnection

当新连接到来时，便会将新到的连接封转成TcpConnection丢入子线程中执行

private:
	// 四种状态 ｛未连接，正在连接，已经连接，断开连接中｝
    enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting};
    void setState(StateE state) { state_ = state; }
	
	// 所属loop(子线程)
    EventLoop* loop_;
    const std::string name_;
    std::atomic_int state_;
    bool reading_;
	
	// 绑定的socket，channel
    std::unique_ptr socket_;
    std::unique_ptr channel_;
	
	// 本地地址与对面地址
    const InetAddress localAddr_;
    const InetAddress peerAddr_;
	
	// 各种回调
    ConnectionCallback connectionCallback_;
    MessageCallback messageCallback_;
    WriteCompleteCallback writeCompleteCallback_;
    HighWaterMarkCallback highWaterMarkCallback_;
    CloseCallback closeCallback_;

    size_t highWaterMark_;
	
	// 自定义buffer(用户空间)
    Buffer inputBuffer_;
    Buffer outputBuffer_;
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

当connection构建时，调用connectEstableished方法，表示连接建立完成

// 连接建立
void TcpConnection::connectEstableished() {
    state_ = kConnected;  // 设置状态
    channel_->tie(shared_from_this()); // 绑定
    channel_->enableReading(); // 设置可读
    connectionCallback_(shared_from_this());  // 调用连接建立回调，由用户提供
}
1
2
3
4
5
6
7

同样的，在连接销毁时，调用connectDestroyed方法，将资源销毁

// 连接销毁 
void TcpConnection::connectDestroyed() {
    if(state_ == kConnected) {
        setState(kDisconnected);
        channel_->disableAll();
        connectionCallback_(shared_from_this());
    }
    channel_->remove();
}
1
2
3
4
5
6
7
8
9

​ TcpConnection在构造函数中设置四种回调，分别是读写回调，错误回调以及关闭回调，在对应情况下会分别执行

channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));
1
2
3
4

handleRead

void TcpConnection::handleRead(Timestamp receiveTime) {
    int saveError = 0;
    // 从socket中将内容从内核空间读入用户空间的buffer
    int n = inputBuffer_.readFd(channel_->fd(), &saveError);
    // 有内容读入，调用消息回调
    if(n > 0) {
        // messageCallback由用户自定义，代表服务器执行的功能，从TcpServer传入connection
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    // 无内容，关闭连接
    else if(n == 0) {
        handleClose();
    }
    // 发生错误，报错
    else {
        errno = saveError;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

handleWrite

void TcpConnection::handleWrite() {
    // 如果channel为写事件，则将用户空间中缓存区内容复制给内核空间中的socket进行发送
    if(channel_->isWriteEvent()) {
        int saveError = 0;
        // 复制内容
        int n = outputBuffer_.writeFd(channel_->fd(), &saveError);
        if(n > 0) {
            outputBuffer_.retrieve(n);
            if(outputBuffer_.readableBytes() == 0) {
                channel_->disableWriting();
                // 写完成后，调用写完成回调
                if(writeCompleteCallback_) {
                    loop_->queueInLoop(
                        std::bind(writeCompleteCallback_, shared_from_this())
                    );
                }
            }
            if(state_ == kDisconnecting) {
                shutdownInLoop();
            }
        }
        else {
            LOG_ERROR("TcpConnection::handleWrite");
        }
    }
    else {
        LOG_ERROR("TcpConnection fd=%d is down, no more writing \n", channel_->fd());
    }
}
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

待消息复制后，可调用send方法发送消息(由用户调用)

注意发送时用户空间缓冲区

void TcpConnection::send(const std::string &buf) {
    if(state_ == kConnected) {
        if(loop_->isInLoopThread()) {
            sendInLoop(buf.c_str(), buf.size());
        }
        else {
            loop_->runInLoop(
                std::bind(&TcpConnection::sendInLoop, this, buf.c_str(), buf.size())
            );
        }
    }
}

void TcpConnection::sendInLoop(const void *message, size_t len) {
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;

    // 判断连接状态
    if(state_ == kDisconnected) {
        LOG_ERROR("disconnected, give up writing!");
        return ;
    }
    // 第一次发送数据，且缓冲区没有待发送数据
    if(!channel_->isWriteEvent() && outputBuffer_.readableBytes() == 0) {
        nwrote = ::write(channel_->fd(), message, len);
        if(nwrote >= 0) {
            remaining = len - nwrote;
            // 一次性发送完成，调用写完成事件
            if(remaining == 0 && writeCompleteCallback_) {
                loop_->queueInLoop(
                    std::bind(writeCompleteCallback_, shared_from_this())
                );
            }
        }
        else {
            nwrote = 0;
            if(errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop");
                if(errno == EPIPE || errno == ECONNRESET) { //对方关闭管道或连接重置
                    faultError = true;
                }
            }
        }
    }

    // 还有剩余数据未发送，将剩余数据保存到输出buffer中，并给channel注册epollout事件
    // poller发现发送缓冲区有空间，通知对应的channel调用handlewrite事件
    if(!faultError && remaining > 0) { 
        size_t oldlen = outputBuffer_.readableBytes();
        if(oldlen + remaining >= highWaterMark_ && oldlen < highWaterMark_ && highWaterMarkCallback_) {
            loop_->queueInLoop(
                std::bind(highWaterMarkCallback_, shared_from_this(), oldlen+remaining)
            );
        }
        outputBuffer_.append((char*)message+nwrote, remaining);
        if(!channel_->isWriteEvent()) {
            channel_->enableWriting();
        }
    }

}
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

handleClose与handleError

void TcpConnection::handleClose() {
    LOG_INFO("fd=%d, state=%d \n", channel_->fd(), (int)state_);
    setState(kDisconnected);

    TcpConnectionPtr connPtr(shared_from_this());
    connectionCallback_(connPtr);
    closeCallback_(connPtr);
}

void TcpConnection::handleError() {
    int optval;
    socklen_t optlen = sizeof optval;
    int err = 0;
    if(::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
        err = errno;
    }
    else {
        err = optval;
    }
    LOG_ERROR("TcpConnection::handleError name:%s - SO_ERROR:%d \n", name_.c_str(), err);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21