简介
TcpConnection
用于管理一个具体的 TCP 连接，比如消息的接收与发送，完成用户指定的连接回调 connectionCallback。
TcpConnection 有四个状态，简单的状态图：
成员及属性解析
主要接口
send
发送数据的主要接口，最终通过内部实现在runInLoop中发送数据
回调setter
connectionEstablished
当连接建立时，应当只执行一次
将自身的shared_from_this指针与Channel绑定
令Channel激活对可读IO事件的关注
connectionDestroyed
当连接断开时，应当只执行一次
将自己的Channel从所属EventLoop中移除
主要成员
loop
主要回调都通过EventLoop所在线程处理
Channel
通过Channel的回调调用自己的回调
Socket
连接所属的套接字fd
localaddr，peeraddr
本地和对端socketaddr
各种回调callback
inputbuffer，outputbuffer
应用层输入，输出缓冲区
源码剖析
源码已经编写注释
TcpConnection.h
// Copyright 2010, Shuo Chen. All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
// Author: Shuo Chen (chenshuo at chenshuo dot com)
//
// This is a public header file, it must only include public header files.
#ifndef MUDUO_NET_TCPCONNECTION_H
#define MUDUO_NET_TCPCONNECTION_H
#include "muduo/base/noncopyable.h"
#include "muduo/base/StringPiece.h"
#include "muduo/base/Types.h"
#include "muduo/net/Callbacks.h"
#include "muduo/net/Buffer.h"
#include "muduo/net/InetAddress.h"
#include <memory>
#include <boost/any.hpp>
// struct tcp_info is in <netinet/tcp.h>
struct tcp_info;
namespace muduo
{
namespace net
{
class Channel;
class EventLoop;
class Socket;
///
/// TCP connection, for both client and server usage.
///
/// This is an interface class, so don't expose too much details.
/**
* TcpServer => Acceptor => 有一个新用户连接，通过accept函数拿到connfd
* =》 TcpConnection 设置回调 =》 Channel =》 Poller =》 Channel的回调操作
*
*/
class TcpConnection : noncopyable,
public std::enable_shared_from_this<TcpConnection>
{
public:
/// Constructs a TcpConnection with a connected sockfd
///
/// User should not create this object.
TcpConnection(EventLoop* loop,
const string& name,
int sockfd,
const InetAddress& localAddr,//存储自己的addr信息
const InetAddress& peerAddr);//存储对方的addr信息
~TcpConnection();
EventLoop* getLoop() const {
return loop_; }
const string& name() const {
return name_; }
const InetAddress& localAddress() const {
return localAddr_; }
const InetAddress& peerAddress() const {
return peerAddr_; }
bool connected() const {
return state_ == kConnected; }
bool disconnected() const {
return state_ == kDisconnected; }
// return true if success.
bool getTcpInfo(struct tcp_info*) const;
string getTcpInfoString() const;
// void send(string&& message); // C++11
void send(const void* message, int len);
void send(const StringPiece& message);
// void send(Buffer&& message); // C++11
void send(Buffer* message); // this one will swap data
void shutdown(); // NOT thread safe, no simultaneous calling
// void shutdownAndForceCloseAfter(double seconds); // NOT thread safe, no simultaneous calling
void forceClose();
void forceCloseWithDelay(double seconds);
void setTcpNoDelay(bool on);
// reading or not
void startRead();
void stopRead();
bool isReading() const {
return reading_; }; // NOT thread safe, may race with start/stopReadInLoop
/* 设置TCP上下文 */
void setContext(const boost::any& context)
{
context_ = context; }
/* 获取TCP上下文 */
const boost::any& getContext() const
{
return context_; }
boost::any* getMutableContext()
{
return &context_; }
void setConnectionCallback(const ConnectionCallback& cb)
{
connectionCallback_ = cb; }
void setMessageCallback(const MessageCallback& cb)
{
messageCallback_ = cb; }
void setWriteCompleteCallback(const WriteCompleteCallback& cb)
{
writeCompleteCallback_ = cb; }
/* 设置高水位回调函数和高水位值，当缓冲区的size达到highWaterMark时触发此请求 */
void setHighWaterMarkCallback(const HighWaterMarkCallback& cb, size_t highWaterMark)
{
highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; }
/// Advanced interface
Buffer* inputBuffer()
{
return &inputBuffer_; }
Buffer* outputBuffer()
{
return &outputBuffer_; }
/// Internal use only.
void setCloseCallback(const CloseCallback& cb)
{
closeCallback_ = cb; }
// called when TcpServer accepts a new connection
// 连接建立
void connectEstablished(); // should be called only once
// called when TcpServer has removed me from its map
// 连接建立
void connectDestroyed(); // should be called only once
private:
/*
kDisconnected 已经断开连接
kConnecting 正在连接
kConnected 已经连接
kDisconnecting 正在断开连接
*/
enum StateE {
kDisconnected, kConnecting, kConnected, kDisconnecting };
void handleRead(Timestamp receiveTime);
void handleWrite();
void handleClose();
void handleError();
// void sendInLoop(string&& message);
void sendInLoop(const StringPiece& message);
void sendInLoop(const void* message, size_t len);
void shutdownInLoop();
// void shutdownAndForceCloseInLoop(double seconds);
void forceCloseInLoop();
void setState(StateE s) {
state_ = s; }
const char* stateToString() const;
void startReadInLoop();
void stopReadInLoop();
EventLoop* loop_;//subloop
const string name_;
StateE state_; // FIXME: use atomic variable,连接状态
bool reading_;//监听read标志
// we don't expose those classes to client.
//每一个client connect都有一个socket fd和这个fd的channel
std::unique_ptr<Socket> socket_;
std::unique_ptr<Channel> channel_;
const InetAddress localAddr_;//当前主机的地址信息
const InetAddress peerAddr_;//对端主机地址信息
ConnectionCallback connectionCallback_;
MessageCallback messageCallback_;
WriteCompleteCallback writeCompleteCallback_;
HighWaterMarkCallback highWaterMarkCallback_;
CloseCallback closeCallback_;
size_t highWaterMark_;//存储高水位变量
Buffer inputBuffer_;//接收数据的buffer
Buffer outputBuffer_; //发送数据的buffer FIXME: use list<Buffer> as output buffer.
boost::any context_;
// FIXME: creationTime_, lastReceiveTime_
// bytesReceived_, bytesSent_
};
typedef std::shared_ptr<TcpConnection> TcpConnectionPtr;
} // namespace net
} // namespace muduo
#endif // MUDUO_NET_TCPCONNECTION_H
TcpConnection.cc
// Copyright 2010, Shuo Chen. All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.
// Author: Shuo Chen (chenshuo at chenshuo dot com)
#include "muduo/net/TcpConnection.h"
#include "muduo/base/Logging.h"
#include "muduo/base/WeakCallback.h"
#include "muduo/net/Channel.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/Socket.h"
#include "muduo/net/SocketsOps.h"
#include <errno.h>
using namespace muduo;
using namespace muduo::net;
/* 默认的当连接建立和断开时的回调函数 */
void muduo::net::defaultConnectionCallback(const TcpConnectionPtr& conn)
{
LOG_TRACE << conn->localAddress().toIpPort() << " -> "
<< conn->peerAddress().toIpPort() << " is "
<< (conn->connected() ? "UP" : "DOWN");
// do not call conn->forceClose(), because some users want to register message callback only.
}
/* 默认的收消息的回调函数 */
void muduo::net::defaultMessageCallback(const TcpConnectionPtr&,
Buffer* buf,
Timestamp)
{
buf->retrieveAll();//直接把消息全扔掉
}
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)//高水位设置64mb
{
channel_->setReadCallback(
std::bind(&TcpConnection::handleRead, this, _1));
channel_->setWriteCallback(
std::bind(&TcpConnection::handleWrite, this));
channel_->setCloseCallback(
std::bind(&TcpConnection::handleClose, this));
channel_->setErrorCallback(
std::bind(&TcpConnection::handleError, this));
LOG_DEBUG << "TcpConnection::ctor[" << name_ << "] at " << this
<< " fd=" << sockfd;
socket_->setKeepAlive(true);//设置保持这个fd活跃,类似心跳包,不过间隔是2h(很长)
}
TcpConnection::~TcpConnection()
{
LOG_DEBUG << "TcpConnection::dtor[" << name_ << "] at " << this
<< " fd=" << channel_->fd()
<< " state=" << stateToString();
assert(state_ == kDisconnected);
}
bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const//获取关于这个fd的相关tcp信息
{
return socket_->getTcpInfo(tcpi);
}
string TcpConnection::getTcpInfoString() const//将fd的相关tcp信息格式化为字符串
{
char buf[1024];
buf[0] = '\0';
socket_->getTcpInfoString(buf, sizeof buf);
return buf;
}
void TcpConnection::send(const void* data, int len)
{
send(StringPiece(static_cast<const char*>(data), len));
}
void TcpConnection::send(const StringPiece& message)
{
if (state_ == kConnected)
{
//如果调用该函数的线程与subloop是同一个线程,则直接调用,否则调用runInLoop将事件加入subloop的处理队列
if (loop_->isInLoopThread())
{
sendInLoop(message);
}
else
{
void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
loop_->runInLoop(
std::bind(fp,
this, // FIXME
message.as_string()));
//std::forward<string>(message)));
}
}
}
// FIXME efficiency!!!
void TcpConnection::send(Buffer* buf)
{
if (state_ == kConnected)
{
//如果调用该函数的线程与subloop是同一个线程,则直接调用,否则调用runInLoop将事件加入subloop的处理队列
if (loop_->isInLoopThread())
{
sendInLoop(buf->peek(), buf->readableBytes());
buf->retrieveAll();
}
else
{
void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
loop_->runInLoop(
std::bind(fp,
this, // FIXME
buf->retrieveAllAsString()));
//std::forward<string>(message)));
}
}
}
void TcpConnection::sendInLoop(const StringPiece& message)
{
sendInLoop(message.data(), message.size());
}
//发送数据
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发送
// 说明当前这一次write，并没有把数据全部发送出去，剩余的数据需要保存到缓冲区当中，然后给channel
// 注册epollout事件，poller发现tcp的发送缓冲区有空间，会通知相应的sock-channel，调用writeCallback_回调方法
// 也就是调用TcpConnection::handleWrite方法，把发送缓冲区中的数据全部发送完成
if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
{
nwrote = sockets::write(channel_->fd(), data, len);
if (nwrote >= 0)
{
remaining = len - nwrote;//刷新剩余未发送的数据
// 既然在这里数据全部发送完成，就不用再给channel设置epollout事件了
//如果数据全部发送完了,并且注册了发送完毕的回调函数,则调用
if (remaining == 0 && writeCompleteCallback_)
{
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);
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);//将尚未发送的数据加入缓冲区
if (!channel_->isWriting())
{
//监控写事件,当缓冲区可写时subloop会调用handleWrite()函数将剩下的写入,写完后会关闭可写事件的监控
channel_->enableWriting();
}
}
}
void TcpConnection::shutdown()//关闭读写通道,并设置标志为断开连接
{
// FIXME: use compare and swap
if (state_ == kConnected)
{
setState(kDisconnecting);//将状态设置为正在关闭
// FIXME: shared_from_this()?
loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
}
}
void TcpConnection::shutdownInLoop()//关闭socket fd写通道
{
loop_->assertInLoopThread();
if (!channel_->isWriting())//判断是否监听了写事件
{
// we are not writing
socket_->shutdownWrite();//关闭写通道
}
}
// void TcpConnection::shutdownAndForceCloseAfter(double seconds)
// {
// // FIXME: use compare and swap
// if (state_ == kConnected)
// {
// setState(kDisconnecting);
// loop_->runInLoop(std::bind(&TcpConnection::shutdownAndForceCloseInLoop, this, seconds));
// }
// }
// void TcpConnection::shutdownAndForceCloseInLoop(double seconds)
// {
// loop_->assertInLoopThread();
// if (!channel_->isWriting())
// {
// // we are not writing
// socket_->shutdownWrite();
// }
// loop_->runAfter(
// seconds,
// makeWeakCallback(shared_from_this(),
// &TcpConnection::forceCloseInLoop));
// }
void TcpConnection::forceClose()//强制关闭
{
// FIXME: use compare and swap
if (state_ == kConnected || state_ == kDisconnecting)
{
setState(kDisconnecting);
loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
}
}
void TcpConnection::forceCloseWithDelay(double seconds)//强制延迟seconds关闭
{
if (state_ == kConnected || state_ == kDisconnecting)
{
setState(kDisconnecting);
loop_->runAfter(
seconds,
makeWeakCallback(shared_from_this(),
&TcpConnection::forceClose)); // not forceCloseInLoop to avoid race condition
}
}
void TcpConnection::forceCloseInLoop()//强制关闭
{
loop_->assertInLoopThread();
if (state_ == kConnected || state_ == kDisconnecting)
{
// as if we received 0 byte in handleRead();
handleClose();
}
}
const char* TcpConnection::stateToString() const//将连接状态转化为字符串
{
switch (state_)
{
case kDisconnected:
return "kDisconnected";
case kConnecting:
return "kConnecting";
case kConnected:
return "kConnected";
case kDisconnecting:
return "kDisconnecting";
default:
return "unknown state";
}
}
void TcpConnection::setTcpNoDelay(bool on)//禁用nagle算法
{
socket_->setTcpNoDelay(on);
}
void TcpConnection::startRead()//开始监听读事件
{
loop_->runInLoop(std::bind(&TcpConnection::startReadInLoop, this));
}
void TcpConnection::startReadInLoop()//开始监听读事件
{
loop_->assertInLoopThread();
if (!reading_ || !channel_->isReading())
{
channel_->enableReading();
reading_ = true;
}
}
void TcpConnection::stopRead()//停止监听读事件
{
loop_->runInLoop(std::bind(&TcpConnection::stopReadInLoop, this));
}
void TcpConnection::stopReadInLoop()//停止监听读事件
{
loop_->assertInLoopThread();
if (reading_ || channel_->isReading())
{
channel_->disableReading();
reading_ = false;
}
}
//在TcpServer->Acceptor接收到connect时候调用
void TcpConnection::connectEstablished()
{
loop_->assertInLoopThread();
assert(state_ == kConnecting);
setState(kConnected);//设置连接状态为完成连接
channel_->tie(shared_from_this()); //将this给weak_ptr传给channel监控存活状态
channel_->enableReading();//监听读事件
connectionCallback_(shared_from_this());
}
void TcpConnection::connectDestroyed()
{
loop_->assertInLoopThread();
if (state_ == kConnected)
{
setState(kDisconnected);//设置连接状态为完成关闭
channel_->disableAll();//关闭所有事件的监听
connectionCallback_(shared_from_this());
}
channel_->remove();//删除channel
}
void TcpConnection::handleRead(Timestamp receiveTime)
{
loop_->assertInLoopThread();
int savedErrno = 0;
ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);//将内核read缓冲区的数据读到input缓冲区
if (n > 0)
{
messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);//调用接收信息的回调函数
}
else if (n == 0)//表示对方已经断开socket connect
{
handleClose();
}
else
{
errno = savedErrno;
LOG_SYSERR << "TcpConnection::handleRead";
handleError();
}
}
void TcpConnection::handleWrite()
{
loop_->assertInLoopThread();
if (channel_->isWriting())//判断是否监听了写事件
{
//将outputBuffer的数据写入内核发送缓冲区
ssize_t n = sockets::write(channel_->fd(),
outputBuffer_.peek(),
outputBuffer_.readableBytes());
if (n > 0)
{
outputBuffer_.retrieve(n);//调整buffer的下标
if (outputBuffer_.readableBytes() == 0)//表示已经全部写完了,缓冲区没有数据可以发送了
{
channel_->disableWriting();//关闭写时间的监测
if (writeCompleteCallback_)//调用写完毕的回调函数
{
loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
}
//如果之前设置了关闭连接的标志,那么现在可以关闭写端了
//(如果之前在调用关闭连接时候,还在发送数据,就要等发送完毕了在关闭)
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";
}
}
// poller => channel::closeCallback => TcpConnection::handleClose
void TcpConnection::handleClose()
{
loop_->assertInLoopThread();
LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
assert(state_ == kConnected || state_ == kDisconnecting);
// we don't close fd, leave it to dtor, so we can find leaks easily.
setState(kDisconnected);
channel_->disableAll();//不在关心任何事件
TcpConnectionPtr guardThis(shared_from_this());
connectionCallback_(guardThis);// 执行连接关闭的回调
// must be the last line
closeCallback_(guardThis); // 关闭连接的回调 执行的是TcpServer::removeConnection回调方法
}
void TcpConnection::handleError()
{
int err = sockets::getSocketError(channel_->fd());
LOG_ERROR << "TcpConnection::handleError [" << name_
<< "] - SO_ERROR = " << err << " " << strerror_tl(err);
}
总结
TcpConnection 构造时接收参数有 TCP 连接的 sockfd，服务端地址 localAddr，客户端地址 peerAddr，并通过 Socket 封装 sockfd。并用 Channel 管理该 sockfd，向 Channel 注册可读、可写、关闭、出错回调函数，用于 Poller 返回就绪事件后 Channel::handleEvent() 执行相应事件的回调。
TcpConnection::handleRead()：当连接对应的 sockfd 有可读事件发生时调用，主要是将数据读到 Buffer 中，执行消息回调函数 messageCallback_()。_TcpConnection::handleWrite()：当连接对应的 sockfd 有可写事件发生时调用，主要是将 Buffer 中的数据发送出去，如果一次性发送完毕，则执行用户指定的回调 writeCompleteCallback_()，若一次没有发送完， muduo 采用 LT 模式， 会反复触发可写事件，下次还有机会发送剩下的数据。send一系列函数是可以用户或者其他线程调用，用于发送信息。如果不是在IO线程，它会把实际工作转移到IO线程调用。首先检查 TcpConnection 对应的 Socket 是否注册了可写事件，若注册了可写事件表明输出缓冲区 outputBuffer中已经有数据等待发送，为了保证不乱序，这次的数据只要追加到输出缓冲区中，通过 Channel::handleEvent() -> TcpConnection::handleWrite() 来发送。如果Socket 没有注册可写事件，输出缓冲区没有数据，那么这次的消息可以直接通过 write 发送，如果没有一次性发送完毕，那么 message 剩余的数据仍然要 append 到 outputBuffer 中，并向 Poller 注册可写事件，当 socket 变得可写时，Channel 会调用 TcpConnection::handleWrite() 来发送 outputBuffer 中堆积的数据，发送完毕后立刻停止监听可写事件，避免 busy loop。无论是 sendInLoop() -> write() 还是 Channel::handleEvent() -> handleWrite()，只要确定发送完 message 或者 outputBuffer 中的数据，那么都要调用用户指定的回调 writeCompleteCallback()。
使用 epoll 的 LT 模式，当 socket 可写时，会不停的触发 socket 的可写事件，这个时候如何解决？
第一种方式：需要向 socket 写数据时，注册此 socket 的可写事件，接收到可写事件后，然后调用 write/send 写数据，当所有数据都写完后，立刻停止观察 writable 事件，避免 busy loop。
第二种，需要发送数据时，直接调用 write/send 写，如果没有发送完，那么开是监听此 socket 的 writable 事件，然后接收到可写事件后，调用 write/send 发送数据，当所有数据都写完后，立刻停止观察 writable 事件，避免 busy loop。 muduo 采用的 LT 模式，就是用的第二种解决这个问题。
此外 TcpConnection 还有几个小功能，比如 TcpConnection::setTcpNoDelay() 和 TcpConnection::setKeepAlive()。TCP No Delay 和 TCP keepalive 都是常用的 TCP 选项，前者的作用是禁止 Nagle 算法，避免连续发包出现延迟，这对编写低延迟网络服务很重要。后者的作用是定期探查 TCP 连接是否还存在，一般来说如果有应用层心跳的话，TCP keepalive 不是必须的。并且时间很长(默认2h)
leEvent() -> handleWrite()，只要确定发送完 message 或者 outputBuffer_ 中的数据，那么都要调用用户指定的回调 writeCompleteCallback()。
使用 epoll 的 LT 模式，当 socket 可写时，会不停的触发 socket 的可写事件，这个时候如何解决？
第一种方式：需要向 socket 写数据时，注册此 socket 的可写事件，接收到可写事件后，然后调用 write/send 写数据，当所有数据都写完后，立刻停止观察 writable 事件，避免 busy loop。
第二种，需要发送数据时，直接调用 write/send 写，如果没有发送完，那么开是监听此 socket 的 writable 事件，然后接收到可写事件后，调用 write/send 发送数据，当所有数据都写完后，立刻停止观察 writable 事件，避免 busy loop。 muduo 采用的 LT 模式，就是用的第二种解决这个问题。
此外 TcpConnection 还有几个小功能，比如 TcpConnection::setTcpNoDelay() 和 TcpConnection::setKeepAlive()。TCP No Delay 和 TCP keepalive 都是常用的 TCP 选项，前者的作用是禁止 Nagle 算法，避免连续发包出现延迟，这对编写低延迟网络服务很重要。后者的作用是定期探查 TCP 连接是否还存在，一般来说如果有应用层心跳的话，TCP keepalive 不是必须的。并且时间很长(默认2h)
