TcpConnection的读写操作【深度剖析】

---

前言

今天总结TcpConnection类的读写事件。

一、TcpConnection的读

当Poller检测到套接字的Channel处于可读状态时，会调用Channel的回调函数，回调函数中根据不同激活原因调用不同的函数，这些函数都由TcpConnection在创建Channel之初提供，当可读时，调用TcpConnection的可读函数handleRead，而在这个函数中，读缓冲区就会从内核的tcp缓冲区读取数据。

void TcpConnection::handleRead(Timestamp receiveTime)
{
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0)
    {
        // 已建立连接的用户，有可读事件发生了，调用用户传入的回调操作onMessage
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        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
21

TcpConnection::handleRead( )函数首先调用Buffer_.readFd(channel_->fd(), &saveErrno)，该函数底层调用Linux的函数readv( )，将Tcp接收缓冲区数据拷贝到用户定义的缓冲区中（inputBuffer_）。如果在读取拷贝的过程中发生了什么错误，这个错误信息就会保存在savedErrno中。
对于缓冲区 Buffer::readFd()函数之前的文章已经剖析过了。

二、TcpConnection的写

TcpConnection::send() 方法是用户调用发送接口，会调用TcpConnection::sendInLoop() 方法来处理具体的发送操作。如果在当前线程直接发送，就会调用 sendInLoop() 方法处理，否则需要把发送任务加入到事件循环中，等待对应的线程处理。

//给用户提供的 发送接口
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()
            ));
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在TcpConnection::sendInLoop() 方法的实现中接通过系统调用 write() 发送数据，如果有剩余未发送的数据，则会将数据添加到发送缓冲区中，并注册 channel 的可写事件，等待事件循环通知空闲后再进行发送。

/**
 * 发送数据  应用写的快， 而内核发送数据慢， 需要把待发送数据写入缓冲区，
 *  而且设置了水位回调
 */ 
void TcpConnection::sendInLoop(const void* data, size_t len)
{
    ssize_t nwrote = 0;
    // remaining是没发送完的数据
    size_t remaining = len;
    bool faultError = false;

    // 之前调用过该connection的shutdown，不能再进行发送了
    if (state_ == kDisconnected)
    {
        LOG_ERROR("disconnected, give up writing!");
        return;
    }

    // 表示channel_第一次开始写数据，而且缓冲区没有待发送数据
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {

        // 返回的是具体发送的 数据
        nwrote = ::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            // 如果放松完了 ，并且注册了 发送完回调函数
            if (remaining == 0 && writeCompleteCallback_)
            {
                // 既然在这里数据全部发送完成，就不用再给channel设置epollout事件了
                loop_->queueInLoop(
                    std::bind(writeCompleteCallback_, shared_from_this())
                );
            }
        }
        else // nwrote < 0
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_ERROR("TcpConnection::sendInLoop");
                if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE  RESET
                {
                    faultError = true;
                }
            }
        }
    }

    // 说明当前这一次write，并没有把数据全部发送出去，
    // 剩余的数据需要保存到缓冲区当中，然后给channel
    // 注册epollout事件，poller发现tcp的发送缓冲区有空间，
    // 因为是lt模式  如果缓存区空余 就会不断地提醒
    // 会通知相应的sock-channel，调用writeCallback_回调方法也就是hanldwrite方法
    // 也就是调用TcpConnection::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*)data + nwrote, remaining);
        if (!channel_->isWriting())
        {
            // 这里一定要注册channel的写事件，否则poller不会给channel通知epollout
            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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

发送缓冲区中有数据时，TcpConnection::handleWrite() 方法会被调用来处理具体的发送操作。在该方法中，首先会判断 channel 是否可写，如果可写则通过系统调用 writeFd() 将发送缓冲区中的数据写入到套接字中。如果写入成功，就会从发送缓冲区中删除已经发送的数据，并判断是否还有剩余数据，如果没有，则禁用 channel 的写事件，并执行可写回调函数。如果还有剩余数据，则会继续等待事件循环通知空闲后再次进行发送。

// 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{
    if (channel_->isWriting())
    {
        int savedErrno = 0;
        ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);
        if (n > 0)
        {
            // 有数据发送成功  n个数据已经处理过了 把readable 向右移
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                // 已经发送完成了 编程不可写   执行回调写完回调writeCompleteCallback_
                channel_->disableWriting();
                if (writeCompleteCallback_)
                {
                    // 唤醒loop_对应的thread线程，执行回调
                    // 唤醒线程 执行写完之后的回调事件
                    loop_->queueInLoop(
                        std::bind(writeCompleteCallback_, shared_from_this())
                    );
                }
                if (state_ == kDisconnecting)
                {
                    // 如果还有数据但是 就调用了 shutdown
                    //   state_就变成了 == kDisconnecting
                    // 但是 需要等待 数据传输完成 再调用shutdownInLoop
                    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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

这里的细节问题就是如果想要关闭连接，那么通常是先关闭读端，等到将写缓冲区所有数据都写到tcp缓冲区后，再关闭写端，否则这些数据就不能发送给对端了

三、TcpConnection的关闭

需要关闭时候setState(kDisconnecting);把状态设置为kDisconnecting但是没有立即关闭，而是判断是否还有数据可写。

// 关闭连接
void TcpConnection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        loop_->runInLoop(
            std::bind(&TcpConnection::shutdownInLoop, this)
        );
    }
}
void TcpConnection::shutdownInLoop()
{
    // 如果buffer还有数据，这个就是writing状态
    // 会一直发， 知道发完 然后监控到状态是kDisconnecting 
    // 再次调用这个函数 ，就会关闭了
    // 保证数据发送完
    if (!channel_->isWriting()) // 说明outputBuffer中的数据已经全部发送完成
    {
        socket_->shutdownWrite(); // 关闭写端
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

如果buffer还有数据，这个就是writing状态会一直被epoll提醒发送，直到发完 然后监控到状态是kDisconnecting 再次调用这个函数 ，就会关闭了
保证数据发送完。

// 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{
    if (channel_->isWriting())
    {
        int savedErrno = 0;
        ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);
        if (n > 0)
        {
            // 有数据发送成功  n个数据已经处理过了 把readable 向右移
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                // 已经发送完成了 编程不可写   执行回调写完回调writeCompleteCallback_
                channel_->disableWriting();
                if (writeCompleteCallback_)
                {
                    // 唤醒loop_对应的thread线程，执行回调
                    // 唤醒线程 执行写完之后的回调事件
                    loop_->queueInLoop(
                        std::bind(writeCompleteCallback_, shared_from_this())
                    );
                }
                if (state_ == kDisconnecting)
                {
                    // 如果还有数据但是 就调用了 shutdown
                    //   state_就变成了 == kDisconnecting
                    // 但是 需要等待 数据传输完成 再调用shutdownInLoop
                    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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44