handleRead函数
修改TcpConnection::handleRead()代码。现在当某个TcpConnection发生可读事件，调用TcpConnection::handleRead()，先调用inputBuffer.readFd()将内核接收缓冲区数据读取到inputBuffer中，接着调用messageCallback_，用户代码可以按消息界限从inputBuffer_ 中读取数据。
void TcpConnection::handleRead(Timestamp receiveTime)
{
loop_->assertInLoopThread();
int savedErrno = 0;
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();
}
}
send函数
用户代码想要发送数据时，调用TcpConnection::send()，重载了3个版本，都是线程安全的，内部最终都是调用TcpConnection::sendInLoop()。如果不是在当前IO线程调用send时，sendInLoop会在当前IO线程处理doPendingFunctors时被调用。
首先尝试write入内核发送缓冲区，如果内核发送缓冲区满则将未写完的数据添加到outputBuffer中。`只要第一次没写完，下次调用send也会将数据添加到outputBuffer的末尾而不直接write，并关注POLLOUT 事件，当内核发送缓冲区不为满，即发生可写事件，调用TcpConnection::handleWrite() `。
void TcpConnection::sendInLoop(const void *data, size_t len)
{
loop_->assertInLoopThread();
ssize_t nwrote = 0;
size_t remaining = len;
bool error = 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;
// 写完了，回调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) // FIXME: any others?
{
error = true;
}
}
}
}
assert(remaining <= len);
// 没有错误，并且还有未写完的数据（说明内核发送缓冲区满，要将未写完的数据添加到output buffer中）
if (!error && remaining > 0)
{
LOG_TRACE << "I am going to write more data";
size_t oldLen = outputBuffer_.readableBytes();
// 如果超过highWaterMark_（高水位标），回调highWaterMarkCallback_
if (oldLen + remaining >= highWaterMark_
&& oldLen < highWaterMark_
&& highWaterMarkCallback_)
{
loop_->queueInLoop(boost::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
}
outputBuffer_.append(static_cast<const char *>(data) + nwrote, remaining);
if (!channel_->isWriting())
{
channel_->enableWriting(); // 关注POLLOUT事件
}
}
}
handleWrite函数
从outputBuffer_中取出数据写入内核发送缓冲区，当然也许此次并不能完全写入，但只要应用层发送缓冲区不为空，就一直关注POLLOUT事件，当内核发送缓冲区不满时触发再次写入。
// 内核发送缓冲区有空间了，回调该函数
void TcpConnection::handleWrite()
{
loop_->assertInLoopThread();
if (channel_->isWriting())
{
ssize_t n = sockets::write(channel_->fd(),
outputBuffer_.peek(),
outputBuffer_.readableBytes());
if (n > 0)
{
outputBuffer_.retrieve(n);
if (outputBuffer_.readableBytes() == 0) // 应用层发送缓冲区已清空
{
channel_->disableWriting(); // 停止关注POLLOUT事件，以免出现busy loop
if (writeCompleteCallback_) // 回调writeCompleteCallback_
{
// 应用层发送缓冲区被清空，就回调用writeCompleteCallback_
loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
}
if (state_ == kDisconnecting) // 应用层发送缓冲区已清空并且连接状态是kDisconnecting, 要关闭连接
{
shutdownInLoop(); // 关闭连接
}
}
else
{
LOG_TRACE << "I am going to write more data";
}
}
else
{
LOG_SYSERR << "TcpConnection::handleWrite";
// if (state_ == kDisconnecting)
// {
// shutdownInLoop();
// }
}
}
else
{
LOG_TRACE << "Connection fd = " << channel_->fd()
<< " is down, no more writing";
}
}
shutdown函数
shutdown是线程安全地，shutdownInloop保证在IO线程中调用。send函数也是一样。如果在非IO线程调用，它会把message复制一份，传给IO线程的sendInLoop来发送。
如果output buffer里还有待发送的数据，而程序又想关闭连接。对程序而言，调用TcpConnection::send()之后他就认为数据迟早会发出去，那么这时候网络库不能立刻关闭连接，而要等数据发送完毕，Muduo TcpConnection没有提供close，而只提供shutdown，这么做是为了收发数据的完整性。用shutdown而不用close的效果是，如果对方已经发送了数据，这些数据还“在路上”，那么muduo不会漏收这些数据。
如果应用层缓冲区数据还没发完，即还在关注POLLOUT事件，那么shutdown()中只是先设置state_ = kDisconnecting；shutdownInLoop()中判断isWriting()为true，故不会执行shutdownWrite()，handleWrite()函数，当应用层缓冲区数据发完，判断状态为kDisconnecting而且已经disableWriting()，就调用shutdownInLoop() ，此时就会真正关闭写的这一端。
我们发完了数据，于是shutdownWrite，发送TCP FIN分节。对方会读到0字节，然后关闭连接，可读事件发生调用handleRead()，调用handleClose()，进而 调用connectionCallback_，这样客户代码就知道对方断开连接了（判断是否connected()），最后调用closeCallback_（TcpServer::removeConnection()）。
void DaytimeServer::onConnection(const muduo::net::TcpConnectionPtr &conn)
{
if (conn->connected())
{
conn->send(Timestamp::now().toFormattedString() + ”\n”);
conn->shutdown(); // 调用TcpConnection::shutdown()
}
}
void TcpConnection::shutdown()
{
if (state_ == kConnected)
{
setState(kDisconnecting);
// 调用TcpConnection::shutdownInLoop()
loop_->runInLoop(boost::bind(&TcpConnection::shutdownInLoop, this));
}
}
void TcpConnection::shutdownInLoop()
{
loop_->assertInLoopThread();
if (!channel_->isWriting())
{
// we are not writing
socket_->shutdownWrite(); // 调用Socket::shutdownWrite()
}
}
void Socket::shutdownWrite()
{
sockets::shutdownWrite(sockfd_);
}
void sockets::shutdownWrite(int sockfd)
{
int ret = ::shutdown(sockfd, SHUT_WR);
// 检查错误
}
那么muduo什么时候真正close socket 呢？在TcpConnection对象析构的时候。TcpConnection持有一个Socket对象，Socket是一个RAII handler，它的析构函数会close(sockfd_)。
测试程序
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/InetAddress.h>
#include <boost/bind.hpp>
#include <stdio.h>
using namespace muduo;
using namespace muduo::net;
class TestServer
{
public:
TestServer(EventLoop *loop,
const InetAddress &listenAddr)
: loop_(loop),
server_(loop, listenAddr, "TestServer")
{
server_.setConnectionCallback(
boost::bind(&TestServer::onConnection, this, _1));
server_.setMessageCallback(
boost::bind(&TestServer::onMessage, this, _1, _2, _3));
message1_.resize(100);
message2_.resize(200);
std::fill(message1_.begin(), message1_.end(), 'A');
std::fill(message2_.begin(), message2_.end(), 'B');
}
void start()
{
server_.start();
}
private:
void onConnection(const TcpConnectionPtr &conn)
{
if (conn->connected())
{
printf("onConnection(): new connection [%s] from %s\n",
conn->name().c_str(),
conn->peerAddress().toIpPort().c_str());
conn->send(message1_);
conn->send(message2_);
conn->shutdown();
}
else
{
printf("onConnection(): connection [%s] is down\n",
conn->name().c_str());
}
}
void onMessage(const TcpConnectionPtr &conn,
Buffer *buf,
Timestamp receiveTime)
{
muduo::string msg(buf->retrieveAllAsString());
printf("onMessage(): received %zd bytes from connection [%s] at %s\n",
msg.size(),
conn->name().c_str(),
receiveTime.toFormattedString().c_str());
conn->send(msg);
}
EventLoop *loop_;
TcpServer server_;
muduo::string message1_;
muduo::string message2_;
};
int main()
{
printf("main(): pid = %d\n", getpid());
InetAddress listenAddr(8888);
EventLoop loop;
TestServer server(&loop, listenAddr);
server.start();
loop.loop();
}
程序中一旦连接建立，调用onConnection()，send(message1),send(message2)，然后立马shutdown()。由前面分析可知会一直等到outputBuffer_数据全部写到内核发送缓冲区才会真正关闭写端，客户端读到数据后最后read返回0，客户端close导致服务端最终removeConnection。可以看到在handleEvent()处理完毕后TcpConnection才会析构。
WriteCompleteCallback/highWaterMarkCallback
如果我们向一个连接发送send()大流量的数据，发送频率不能太快，因为如果对等方接收不及时，则内核发送缓冲区会堆积数据，根据前面的分析，我们会将数据添加到outputBuffer，导致outputBuffer增长太快。对此可以关注WriteCompleteCallback_，当它被调用时表示outputBuffer_已经被清空，此时再次send()，否则outputBuffer_ 可能一直增长直到撑爆。
可以把WriteCompleteCallback当作是“低水位标”回调函数，highWaterMarkCallback当作是”高水位标“回调函数，即如果对等方接收不及时，outputBuffer会一直增大，`当增长到highWaterMark（具体数值）时，回调highWaterMarkCallback_函数，很可能在函数内主动shutdown。`
TcpClient和TcpServer需要相应地暴露writeCompleteCallback_接口。如果输出缓冲的长度超过用户指定的大小，就会触发回调highWaterMarkCallback_。
下面的程序会不断地发送不同的字符数据，类似chargen协议（DDos）
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/InetAddress.h>
#include <boost/bind.hpp>
#include <stdio.h>
using namespace muduo;
using namespace muduo::net;
class TestServer
{
public:
TestServer(EventLoop *loop,
const InetAddress &listenAddr)
: loop_(loop),
server_(loop, listenAddr, "TestServer")
{
server_.setConnectionCallback(
boost::bind(&TestServer::onConnection, this, _1));
server_.setMessageCallback(
boost::bind(&TestServer::onMessage, this, _1, _2, _3));
server_.setWriteCompleteCallback(
boost::bind(&TestServer::onWriteComplete, this, _1));
// 生成数据
string line;
for (int i = 33; i < 127; ++i)
{
line.push_back(char(i));
}
line += line;
for (size_t i = 0; i < 127 - 33; ++i)
{
message_ += line.substr(i, 72) + '\n';
}
}
void start()
{
server_.start();
}
private:
void onConnection(const TcpConnectionPtr &conn)
{
if (conn->connected())
{
printf("onConnection(): new connection [%s] from %s\n",
conn->name().c_str(),
conn->peerAddress().toIpPort().c_str());
conn->setTcpNoDelay(true);
conn->send(message_);
}
else
{
printf("onConnection(): connection [%s] is down\n",
conn->name().c_str());
}
}
void onMessage(const TcpConnectionPtr &conn,
Buffer *buf,
Timestamp receiveTime)
{
muduo::string msg(buf->retrieveAllAsString());
printf("onMessage(): received %zd bytes from connection [%s] at %s\n",
msg.size(),
conn->name().c_stC r(),
receiveTime.toFormattedString().c_str());
conn->send(msg);
}
void onWriteComplete(const TcpConnectionPtr &conn)
{
conn->send(message_);
}
EventLoop *loop_;
TcpServer server_;
muduo::string message_;
};
int main()
{
printf("main(): pid = %d\n", getpid());
InetAddress listenAddr(8888);
EventLoop loop;
TestServer server(&loop, listenAddr);
server.start();
loop.loop();
}
程序中一旦连接建立就开始send，当outputBuffer_数据全部拷贝到内核发送缓冲区后，回调OnWriteComplete()， 继续send，类似大流量的ddos攻击。
boost::any
TcpConnection中boost::any context_; // 绑定一个未知类型的上下文对象比如HttpContext
可变类型解决方案：void*这种方法不是类型安全的。boost::any任意类型的类型安全存储以及安全的取回。在标准库容器中存放不同类型的方法，比如说vector<boost::any>