muduo网络库代码剖析——TcpConnection类

TcpConnection类的作用:

1.在TcpConnection对象所在线程中（也即那个EventLoop对象所在的线程）监听新连接socetfd的事件

2.设置TcpConnection对象所属的Channel的各种事件的回调函数

3.处理读（接收）事件

4.处理写（发送）事件；是output的数据往socketfd内写（发送）；

5.处理连接关闭事件：涉及到TcpConnection的生命期

服务器端如果主动断开与客户端的连接的流程：

1.如果outputBuffer还有数据没发送完（socket还在关注POLLOUT事件），则只是将连接状态置为kDisconnecting。

2.服务端关闭这端的写，客户端会收到read()返回为0；客户端调用close()

3.服务端会收到两个事件：POLLHUP（对方描述符挂起） POLLIN——在处理的时候是在handleRead()内处理了这个两个事件；

TcpConnection的数据成员：

//TcpConnectio对象和这个EventLoop对象属同一个线程
EventLoop* loop_;

//新连接的名字
const string name_;

//???
StateE state_;  // FIXME: use atomic variable

//????
bool reading_;

//每个新连接的socket对象
// we don't expose those classes to client.
std::unique_ptr<Socket> socket_;

//必然会有个Channel通道来处理新连接后的读写事件
std::unique_ptr<Channel> channel_;

//本地和对端的地址对象
const InetAddress localAddr_;
const InetAddress peerAddr_;

//???
ConnectionCallback connectionCallback_;

//每一次将数据读到inputBuffer时都会调用这个回调函数，对协议包的解析应该在这个回调函数内；
MessageCallback messageCallback_;

//outBuffer缓冲区的数据全部写入到socketfd内核缓冲区以后会调用下面这个函数；
WriteCompleteCallback writeCompleteCallback_;

//当outBuffer缓冲区的大小到达上限值以后会调用下面的回调函数；
HighWaterMarkCallback highWaterMarkCallback_;

//???
CloseCallback closeCallback_;

//用户缓冲区的大小到达一定大小时会做额外的处理：关闭连接。这个是上限值
size_t highWaterMark_;

//从socketfd的读取数据放入的应用层缓冲区
Buffer inputBuffer_;

//将应用层缓冲区数据写入到socketfd发送您给客户端的应用层缓冲区
Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.

//????
boost::any context_;

一、监听新连接socetfd的事件

void TcpConnection::connectEstablished()
{
    loop_->assertInLoopThread();
    assert(state_ == kConnecting);
    setState(kConnected);
    
    //????
    channel_->tie(shared_from_this());
    
    //将这个Channel添加进poll的关注列表内；
    channel_->enableReading();

    //连接建立以后调用，由TcpServer指定，TcpServer又由更上一层指定这个回调函数
    connectionCallback_(shared_from_this());
}

二、设置TcpConnection对象所属的Channel的各种事件的回调函数

TcpConnection::TcpConnection(EventLoop* loop, const string& nameArg, int sockfd, const InetAddress& localAddr, const InetAddress& peerAddr)
  : loop_(CHECK_NOTNULL(loop)),
    name_(nameArg),
    state_(kConnecting),
    reading_(true),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    localAddr_(localAddr),
    peerAddr_(peerAddr),
    
    //设置缓冲区大小的上限值；
    highWaterMark_(64*1024*1024)
{
    //当通道有读事件时候在Channel::handleEvent()内调用：TcpConnection::handleRead()
    channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, _1));
  
    //当通道有写事件的时候在Channel::handleEven()内调用：TcpConnection::handleWrite()
    channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));

    //当通道有关闭事件的时候在Channel::handleEvent()内调用：TcpConnection::handleClose()
    channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));

    //当通道有错误事件的时候在Channel::handleEvent()内调用：TcpConnection::handleError()
    channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));
    LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this << " fd=" << sockfd;

    //开启Tcp/Ip层的心跳包检测
    socket_->setKeepAlive(true);
}

如何借助inputBuffer和outBuffer读取和写入数据到见：

muduo网络库代码剖析——Buffer类

三、处理读（接收）事件

socketfd可读的时候最终会调用下面这个函数将从内核缓冲区的数据读取到应用层inputBuffer缓冲区。如果一次读没有将数据读取完，下一次读同样会调用下面的函数，将数据添加到应用层缓冲区的末尾；

void TcpConnection::handleRead(Timestamp receiveTime)
{
    loop_->assertInLoopThread();
    int savedErrno = 0;

    //将socketfd中的数据从内核的缓冲区读到inpuBuffer的应用层缓冲区;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0)
    {
        //TcpConnection的messageCallback_由TcpServer设置，而TcpServer的又由更上一层（用户应用层）设置。其中如何按协议解包在这个函数内实现。
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        //处理连接关闭事件
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_SYSERR << "TcpConnection::handleRead";
        //处理错误事件
        handleError();
    }
}

四、处理写（发送）事件

重载了多个send函数：

void TcpConnection::send(const void* data, int len)

void TcpConnection::send(const StringPiece& message)

void TcpConnection::send(Buffer* buf)

void TcpConnection::send(Buffer* buf)
{
    if (state_ == kConnected)
    {
        //如果线程是TcpConnection数据成员EventLoop对象所处的线程
        if (loop_->isInLoopThread())
        {
            sendInLoop(buf->peek(), buf->readableBytes());
            
            //对缓冲区复位，readerIndex_和writerIndex_回到最初的位置
            buf->retrieveAll();
        }
        else
        {

            //如果不是当前线程，会转到当前线程调用TcpConnection::sendInLoop()
            void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
  
            //buf->retrieveAllAsString()将数据当成字符串返回???
            loop_->runInLoop(std::bind(fp, this, buf->retrieveAllAsString()));

        }
    }
}

send()函数将outputBuffer的所有数据都拷贝到socketfd内核缓冲区后会调用writeCompleteCallback通知上层发送更多的数据；

大流量应用程序需要关注这个是因为：

如果对等方接收不及时，那么受通告窗口的限制，发送方socketfd内核缓冲区可能不够，那么发送方的发送的数据就会存入应用层缓冲区outputBuffer内。如果持续下去就会导致outputBuffer不断增加。

解决办法就是关注writeCompleteCallback，保证所有数据发送完，再继续发送。

outputBuffer所有数据都拷贝到内核缓冲区后也会调用该writeCompleteCallback；

小流量应用程序不需要关注这个。

void TcpConnection::sendInLoop(const void* data, size_t len)
{
    loop_->assertInLoopThread();
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    
    //如果已经关闭，则返回。
    if (state_ == kDisconnected)
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }

    // if no thing in output queue, try writing directly
    //如果通道（即这个文件描述符）没有关注写事件
    //并且发送缓冲区没有数据。那么可以尝试直接write()
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        nwrote = sockets::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            
            //如果所有数据都发送完了（其实是都拷贝到了内核缓冲区）并且回调函数存在
            if (remaining == 0 && writeCompleteCallback_)
            {
                //交由EventLoop所在的线程去执行；
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
        }
        else // nwrote < 0
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
                {
                    faultError = true;
                }
            }
        }
     }

    assert(remaining <= len);
    
    //如果没有一次性将数据拷贝到内核缓冲区
    //将数据写到outputBuffer的缓冲区
    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(static_cast<const char*>(data)+nwrote, remaining);

        //如果fd没有在关注pollout事件那么设置关注可写事件，这样的话当内核缓冲区有空间可以写了以后
        //会回调TcpConnection::handleWrite()函数；
        if (!channel_->isWriting())
        {
            channel_->enableWriting();
        }
    }
}

如果发送数据写入socketfd时内核缓冲区已满，会将数据写到outputBuffer中。当内核缓冲区可写时会回调TcpConnection::handleWrite()

//内核发送缓冲区有空间了，回调该函数
void TcpConnection::handleWrite()
{
    loop_->assertInLoopThread();
    
    //如果在关注socketfd内核缓冲区的可写事件的话就执行
    if (channel_->isWriting())
    {
        //内核缓冲区有空间，可以直接调用write从应用层缓冲区写到内核缓冲区
        ssize_t n = sockets::write(channel_->fd(), outputBuffer_.peek(), outputBuffer_.readableBytes());
        
        if (n > 0)
        {
            //写入了n个字节就回收应用层的空闲缓冲区空间，实际是下标的移动
            outputBuffer_.retrieve(n);

            //说明应用层发送缓冲区已经清空了
            if (outputBuffer_.readableBytes() == 0)
            {
                //停止关注可写事件（POLLOUT事件）
                channel_->disableWriting();
                //应用层发送缓冲区没有数据了，回调该函数
                if (writeCompleteCallback_)
                {
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }

                //有可能上层应用已经调用了TcpConnection::shutdown()
                //如果outputBuffer还有数据的话会等数据都发送完了才调用关闭连接；这里就是数据发送完的地方了
                if (state_ == kDisconnecting)
                {
                    shutdownInLoop();
                }
            }
        }
        else
        {
            LOG_SYSERR << "TcpConnection::handleWrite";
            // if (state_ == kDisconnecting)
            // {
            //   shutdownInLoop();
            // }
        }
    }
    else
    {
        LOG_TRACE << "Connection fd = " << channel_->fd() << " is down, no more writing";
    }
}

五、处理连接关闭事件：涉及到TcpConnection的生命期

连接关闭的时序图：

TcpConnection生命管理的步骤：

1.在TcpServer新连接到来时用map<string, std:shared_ptr<TcpConnection>>管理TcpConnection对象，此时新建立的TcpConnection引用计数为1；

2.在TcpConnection::connectEstablished()内将TcpConnection对象指针赋值给其Channel成员的std::weak_ptr<void> tie_。由于是弱指针不增加引用计数，依旧为1；

3.客户端选择关闭连接时：TcpConnection的Channel调用的是读事件:在Channel::handleEvent()函数内将weak_ptr弱指针提升为shared_ptr，引用计数为2；

4.再调用TcpConnection::handleRead()，判断读取到的字节为0，再调用TcpConnection::handleClose()，在这里又会赋值一次shared_ptr<TcpConnection>，引用计数为3；

5.再调用TcpConnection的回调函数closeCallback_为TcpServer::removeConnection()，然后再交给EventLoop函数做最后的处理。

6.最后会调用：TcpServer::removeConnectionInLoop()，在这里将第一步加入map容器内的TcpConnection智能指针移除，引用计数为2；

7.在TcpServer::removeConnectionInLoop()内构造了一个std::bind(&TcpConnection::connectDestroyed, conn)新的可调用对象，引用计数为3；

8.其实开始对函数调用进行出栈返回。结束TcpConnection::handleClose()函数，回收栈空间上的一个智能指针，引用计数为2；

9.结束Channel::handleEvent()函数，回收栈空间上的智能指针，引用计数为1，最后一个引用计数会在EventLoop函数内调用完std::bind(&TcpConnection::connectDestroyed, conn)后，TcpConnection对象被释放。