[置顶] Muduo网络库源码分析之TcpConnection Class

connectionCallback

TcpConnection

localAddr

peerAddr

sockfd

Channel::handleEvent()

TcpConnection::connectionCallback

messageCallback

writeCompleteCallback

Channel::handleEvent()

TcpConnection::handle*系列函数

TcpConnection::setConnectionCallback

setCloseCallback

TcpServer::newConncetion()

TcpConnection::set*Callback

TcpConnection::handleRead()

Channel::handleEvent()

handleWrite()

handleClose()

handleError()

TcpConnection::handleRead() -> messageCallback()

Channel::set*Callback

TcpConnection::handleRead()

messageCallback_()

TcpConnection::handleWrite()

writeCompleteCallback_()

TcpConnection::handleClose()

Channel::disableAll()

TcpConnection::closeCallback()

TcpConnection::handleError()

TcpConnection::closeCallback()

TcpServer::removeConnection()

Channel::handleEvent() -> Acceptor::handleRead() -> TcpServer::newConnection()

TcpConnection::setCloseCallback(bind(&TcpServer::removeConnectionCallback, this, _1))

TcpConnection::handleRead()

TcpConnection::handleClose() -> TcpConnection::closeCallback_() -> TcpServer::removeConnection()

TcpConnection::connectEstablishd()

TcpServer::newConnection()

Channel::enableReading()

connectionCallback_()

Channel::handleEvent() -> Acceptor::handleRead() -> TcpServer::newConnection() -> ……->TcpConnection::connectEstablished()

TcpConnection::connectDestroyed()

connectionCallback_()

TcpConnection::handleClose() -> TcpConnection::closeCallback_() -> TcpServer::removeConnection() -> ……->TcpConnection::connectDestroyed()

Channel::handleEvent() -> TcpConnection::handleWrite()

TcpConnection::handleWrite()

sendInLoop() -> write()

Channel::handleEvent() -> handleWrite()

writeCompleteCallback()

TcpConnection::setTcpNoDelay()

TcpConnection::setKeepAlive()

#ifndef MUDUO_NET_TCPCONNECTION_H
#define MUDUO_NET_TCPCONNECTION_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 <boost/any.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/noncopyable.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.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.
class TcpConnection : boost::noncopyable,
public boost::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,
const InetAddress& peerAddr);
~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.
/* 设置TCP关闭的回调函数，仅在内部使用，用于移除持有的TcpConnectionPtr */
void setCloseCallback(const CloseCallback& cb)
{ closeCallback_ = cb; }

// called when TcpServer accepts a new connection
/* TcpServer使用，创建完一个新的连接后调用 */
void connectEstablished();   // should be called only once
// called when TcpServer has removed me from its map
/* TcpServer使用，从 map 中删除掉时调用 */
void connectDestroyed();  // should be called only once

private:
/* TcpConnection 有四种状态，从左到右依次是：已断开 正在连接 已连接 正在断开 */
enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };

/* 处理 connfd 上的事件回调函数 */
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);
/* 通过线程转移操作实现安全关闭TCP连接 */
void shutdownInLoop();
// void shutdownAndForceCloseInLoop(double seconds);
/* 主动关闭连接 */
void forceCloseInLoop();
/* 设置TCP连接的状态 */
void setState(StateE s) { state_ = s; }
const char* stateToString() const;
void startReadInLoop();
void stopReadInLoop();

EventLoop* loop_;     // 处理该TCP连接的EventLoop，该EventLoop内部的epoll监听TCP连接对应的fd
const string name_;   // TCP连接的名字
StateE state_;  // FIXME: use atomic variable     // 本条TCP连接的状态
bool reading_;        //
// we don't expose those classes to client.
boost::scoped_ptr<Socket> socket_;    // TCP连接的fd所在的socket对象，fd的关闭由它决定
boost::scoped_ptr<Channel> channel_;  // TCP连接的fd 对应的 Channel
const InetAddress localAddr_;         // TCP连接本地的地址：ip port
const InetAddress peerAddr_;          // TCP连接对方的地址：ip port
ConnectionCallback connectionCallback_;   // 连接建立时的回调函数
MessageCallback messageCallback_;         // 收到消息时的回调函数
WriteCompleteCallback writeCompleteCallback_;     // 消息写入对方缓冲区时的回调函数
HighWaterMarkCallback highWaterMarkCallback_;     // 高水位回调函数
CloseCallback closeCallback_;         // 关闭TCP连接的回调函数
size_t highWaterMark_;    // 高水位标记
Buffer inputBuffer_;      // TCP连接的输入缓冲区，从连接中读取输入然后存入这里
Buffer outputBuffer_;     // TCP连接的输出缓冲区，要发送的数据保存在这里
// FIXME: use list<Buffer> as output buffer.
boost::any context_;      // TCP连接的上下文，一般用于处理多次消息相互存在关联的情形，例如文件发送
// FIXME: creationTime_, lastReceiveTime_
//        bytesReceived_, bytesSent_
};

/* 使用 shared_ptr 来管理 TCP 连接对象的生存周期 */
typedef boost::shared_ptr<TcpConnection> TcpConnectionPtr;

}
}

#endif  // MUDUO_NET_TCPCONNECTION_H

#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 <boost/bind.hpp>

#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),         // connection 的名字
state_(kConnecting),    // 初始状态是 正在连接
reading_(true),
socket_(new Socket(sockfd)),    // 创建sockfd 对应的Socket
channel_(new Channel(loop, sockfd)), // 创建sockfd对应的Channel
localAddr_(localAddr),
peerAddr_(peerAddr),
highWaterMark_(64*1024*1024)    // 高水位默认是64K
{
/* 注册事件的回调函数，当对应事件发生时会调用 */
channel_->setReadCallback(
boost::bind(&TcpConnection::handleRead, this, _1));
channel_->setWriteCallback(
boost::bind(&TcpConnection::handleWrite, this));
channel_->setCloseCallback(
boost::bind(&TcpConnection::handleClose, this));
channel_->setErrorCallback(
boost::bind(&TcpConnection::handleError, this));
LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
<< " fd=" << sockfd;
socket_->setKeepAlive(true);
}

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
{
return socket_->getTcpInfo(tcpi);
}

string TcpConnection::getTcpInfoString() const
{
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)
{
/* 如果是在loop线程，就直接发送数据 */
if (loop_->isInLoopThread())
{
sendInLoop(message);
}
else
{
/* 否则，转移到loop线程执行 */
loop_->runInLoop(
boost::bind(&TcpConnection::sendInLoop,
this,     // FIXME
message.as_string()));
//std::forward<string>(message)));
}
}
}

// FIXME efficiency!!!
void TcpConnection::send(Buffer* buf)
{
if (state_ == kConnected)
{
if (loop_->isInLoopThread())
{
sendInLoop(buf->peek(), buf->readableBytes());
buf->retrieveAll();
}
else
{
loop_->runInLoop(
boost::bind(&TcpConnection::sendInLoop,
this,     // FIXME
buf->retrieveAllAsString()));
//std::forward<string>(message)));
}
}
}

void TcpConnection::sendInLoop(const StringPiece& message)
{
sendInLoop(message.data(), message.size());
}

/*
* 先尝试直接发送数据，如果一次发送完毕就不会启用WriteCallback
*
* 如果只发送了部分数据，则把剩余数据放入输出缓冲区，
* 并关注writable事件，以后在handleWrite()中发送剩余的数据
*
* 如果当前输出缓冲区已经有待发送的数据，那么就不能先尝试发送了
* 会造成数据乱序
*/
void TcpConnection::sendInLoop(const void* data, size_t len)
{
/* 断言确保是在loop线程 */
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
/* 如果输出缓冲区中没有数据，可以直接往fd写数据，不会乱序 */
if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
{
/* 往fd发送数据 */
nwrote = sockets::write(channel_->fd(), data, len);
/* 发送成功，看是否还有剩余数据 */
if (nwrote >= 0)
{
/* 计算剩余量 */
remaining = len - nwrote;
/* 全部发送完了，执行writeCompleteCallback_回调 */
if (remaining == 0 && writeCompleteCallback_)
{
loop_->queueInLoop(boost::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;
}
}
}
} // end if

assert(remaining <= len);
if (!faultError && remaining > 0)
{
/* 输出缓冲区的剩余字节 */
size_t oldLen = outputBuffer_.readableBytes();
/* 如果现在需要发送的字节数达到高水位，之前没有达到 */
if (oldLen + remaining >= highWaterMark_
&& oldLen < highWaterMark_
&& highWaterMarkCallback_)
{
/* 在 loop 线程中执行高水位回调函数 */
loop_->queueInLoop(boost::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
}

/* 将未发送的data中的数据放入输出缓冲区 */
outputBuffer_.append(static_cast<const char*>(data)+nwrote, remaining);
/* 如果对应的Channel没有在监听write事件 */
if (!channel_->isWriting())
{
/* 注册可写事件 */
channel_->enableWriting();
}
}
}

void TcpConnection::shutdown()
{
// FIXME: use compare and swap
if (state_ == kConnected)     // 执行shutdown必须是在连接正常的情况下
{
setState(kDisconnecting);
// FIXME: shared_from_this()?
loop_->runInLoop(boost::bind(&TcpConnection::shutdownInLoop, this));
}
}

void TcpConnection::shutdownInLoop()
{
loop_->assertInLoopThread();  // 在IO线程中执行
if (!channel_->isWriting())   // 当前没有监听写事件
{
// we are not writing
socket_->shutdownWrite();   // 则关闭写端
}
}
/* 主动关闭连接 */
void TcpConnection::forceClose()
{
// FIXME: use compare and swap
if (state_ == kConnected || state_ == kDisconnecting)
{
setState(kDisconnecting);
loop_->queueInLoop(boost::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
}
}

void TcpConnection::forceCloseWithDelay(double 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(); // 同样调用 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";
}
}

/* 禁用nagle算法，降低网络延迟 */
void TcpConnection::setTcpNoDelay(bool on)
{
socket_->setTcpNoDelay(on);
}

void TcpConnection::startRead()
{
loop_->runInLoop(boost::bind(&TcpConnection::startReadInLoop, this));
}

void TcpConnection::startReadInLoop()
{
loop_->assertInLoopThread();
if (!reading_ || !channel_->isReading())
{
channel_->enableReading();
reading_ = true;
}
}

void TcpConnection::stopRead()
{
loop_->runInLoop(boost::bind(&TcpConnection::stopReadInLoop, this));
}

void TcpConnection::stopReadInLoop()
{
loop_->assertInLoopThread();
if (reading_ || channel_->isReading())
{
channel_->disableReading();
reading_ = false;
}
}

/* 连接建立，TcpServer::newconnection() 调用 */
void TcpConnection::connectEstablished()
{
loop_->assertInLoopThread();
assert(state_ == kConnecting);
setState(kConnected);     // 设置状态为已建立
channel_->tie(shared_from_this());
channel_->enableReading();    // 注册读事件

/* 执行用户建立连接时的逻辑 */
connectionCallback_(shared_from_this());
}

/* 连接关闭，提供给TcpServer使用 TcpServer::removeConnectionInLoop() */
void TcpConnection::connectDestroyed()
{
loop_->assertInLoopThread();
if (state_ == kConnected)
{
setState(kDisconnected);
channel_->disableAll();     // 停止监听所有事件

/* 执行用户的关闭逻辑 */
connectionCallback_(shared_from_this());
}
channel_->remove();   // 从 epoll 中移除connfd
}

/* 处理读事件 */
void TcpConnection::handleRead(Timestamp receiveTime)
{
loop_->assertInLoopThread();
int savedErrno = 0;
/* 从fd读取数据到输入缓冲区中，返回成功读取的字节数 */
ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
if (n > 0)
{
/* 执行消息回调函数 */
messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
}
else if (n == 0)
{
/* 连接关闭，执行关闭回调函数 */
handleClose();
}
else
{
errno = savedErrno;
LOG_SYSERR << "TcpConnection::handleRead";
handleError();
}
}

/* 处理可写事件 往connfd中write */
void TcpConnection::handleWrite()
{
loop_->assertInLoopThread();
/* 如果Channel正在监听 write 事件 */
if (channel_->isWriting())
{
ssize_t n = sockets::write(channel_->fd(),
outputBuffer_.peek(),
outputBuffer_.readableBytes());
if (n > 0)
{
/* 从输出缓冲区中将已经发送的数据删除 */
outputBuffer_.retrieve(n);
/* 所有要发送的数据都发生完毕 */
if (outputBuffer_.readableBytes() == 0)
{
/* 停止监听fd的写事件*/
channel_->disableWriting();
/* 调用用户指定的回调 */
if (writeCompleteCallback_)
{
loop_->queueInLoop(boost::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";
}
}

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();   // Channel停止监听所有事件

TcpConnectionPtr guardThis(shared_from_this());
connectionCallback_(guardThis);   // 执行用户的关闭连接逻辑
// must be the last line
/*
* closeCallback_，绑定到 TcpServer::removeConnection()
* 在里面执行connectDestroyed()
*/
closeCallback_(guardThis);
}

/* 在日志中输出错误信息 */
void TcpConnection::handleError()
{
int err = sockets::getSocketError(channel_->fd());
LOG_ERROR << "TcpConnection::handleError [" << name_
<< "] - SO_ERROR = " << err << " " << strerror_tl(err);
}