muduo网络库学习（六）缓冲区Buffer及TcpConnection的读写操作

tcp

tcp

write/send

tcp

<ip，port>

tcp

poll

read/recv

tcp

write/send

io

io

EAGAIN/EWOULDBLOCK

write/send

io

busy loop

io

tcp

io

tcp

poll

busy loop

poll

tcp

tcp

vector

write/send

read/recv

缓冲区Buffer的设计

muduo

std::vector

vector

Buffer

readerIndex

writerIndex

writeIndex

readerIndex

readerIndex

readerIndex

writerIndex

readerIndex == writerIndex

1KB

8 bytes

readFd

/// A buffer class modeled after org.jboss.netty.buffer.ChannelBuffer
///
/// @code
/// +-------------------+------------------+------------------+
/// | prependable bytes |  readable bytes  |  writable bytes  |
/// |                   |     (CONTENT)    |                  |
/// +-------------------+------------------+------------------+
/// |                   |                  |                  |
/// 0      <=      readerIndex   <=   writerIndex    <=     size
/// @endcode

/*
*
*   缓冲区的设计方法，muduo采用vector连续内存作为缓冲区，libevent则是分块内存
*      1.相比之下，采用vector连续内存更容易管理，同时利用std::vector自带的内存
*        增长方式，可以减少扩充的次数（capacity和size一般不同）
*      2.记录缓冲区数据起始位置和结束位置，写入时写到已有数据的后面，读出时从
*        数据起始位置读出
*      3.起始/结束位置如上图的readerIndex/writeIndex，其中readerIndex为缓冲区
*        数据的起始索引下标，writeIndex为结束位置下标。采用下标而不是迭代器的
*        原因是删除(erase)数据时迭代器可能失效
*      4.开头部分(readerIndex以前)是预留空间，通常只有几个字节的大小，可以用来
*        写入数据的长度，解决粘包问题
*      5.读出和写入数据时会动态调整readerIndex/writeIndex，如果没有数据，二者
*        相等
*/
class Buffer : public muduo::copyable
{
public:
static const size_t kCheapPrepend = 8;
static const size_t kInitialSize = 1024;

explicit Buffer(size_t initialSize = kInitialSize)
: buffer_(kCheapPrepend + initialSize),
readerIndex_(kCheapPrepend),
writerIndex_(kCheapPrepend)
{
assert(readableBytes() == 0);
assert(writableBytes() == initialSize);
assert(prependableBytes() == kCheapPrepend);
}

/* 可读的数据就是起始位置和结束位置中间的部分 */
size_t readableBytes() const
{ return writerIndex_ - readerIndex_; }

size_t writableBytes() const
{ return buffer_.size() - writerIndex_; }

size_t prependableBytes() const
{ return readerIndex_; }

/* 返回数据起始位置 */
const char* peek() const
{ return begin() + readerIndex_; }

/// Read data directly into buffer.
///
/// It may implement with readv(2)
/// @return result of read(2), @c errno is saved
/* 从套接字（内核tcp缓冲区）中读取数据放到读缓冲区中 */
ssize_t readFd(int fd, int* savedErrno);

private:

char* begin()
{ return &*buffer_.begin(); }

const char* begin() const
{ return &*buffer_.begin(); }

private:
/* 缓冲区 */
std::vector<char> buffer_;
/* 数据起始点 */
size_t readerIndex_;
/* 数据结束点 */
size_t writerIndex_;

/* \r\n */
static const char kCRLF[];
};

TcpConnection的读操作

Poller

Channel

Channel

TcpConnection

Channel

TcpConnection

handleRead

tcp

TcpConnection

/*
* 1.TcpConnection构造时，创建一个监听服务器/客户端连接的fd的Channel，设置各种回调函数
* 2.TcpServer设置各种回调函数（可读等），然后调用connectEstablished，将Channel添加到Poller中
* 3.EventLoop继续监听事件，调用Poller
* 4.poll返回，处理激活的Channel，调用Channel的handleEvent
* 5.hanleEvent根据激活事件的类型（可读/可写/挂起/错误）调用不同的处理函数
* 6.若可读，调用hanleRead，TcpConnection中的读缓冲区将内核tcp缓冲区的数据全部读出
* 7.调用用户提供的当可读时执行的回调函数，用户可直接从应用层缓冲区读数据
*/
void TcpConnection::handleRead(Timestamp receiveTime)
{
loop_->assertInLoopThread();
int savedErrno = 0;
/* 读缓冲区从内核tcp中读取数据 */
ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
if (n > 0)
{
/* 如果成功读取数据，调用用户提供的可读时回调函数 */
messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
}
else if (n == 0)
{
/* 如果返回0，说明对端已经close连接，处理close事件，关闭tcp连接 */
handleClose();
}
else
{
/* 出错 */
errno = savedErrno;
LOG_SYSERR << "TcpConnection::handleRead";
handleError();
}
}

TcpConnection

handleRead

readFd

readv(分散读)/writev(集中写)

/*
* 从tcp缓冲区(sockfd)中读取数据，存放到应用层缓冲区中
*   两种情况
*      1.应用层缓冲区足以容纳所有数据
*        直接读取到buffer_中
*      2.应用层缓冲区不够
*        开辟一段栈空间（128k）大小，使用分散读（readv）系统调用读取数据
*        然后为buffer_开辟更大的空间，存放读到栈区的那部分数据
*
*   为什么不在Buffer构造时就开辟足够大的缓冲区
*      1.每个tcp连接都有输入/输出缓冲区，如果连接过多则内存消耗会很大
*      2.防止客户端与服务器端数据交互比较少，造成缓冲区的浪费
*      3.当缓冲区大小不足时，利用vector内存增长的优势，扩充缓冲区
*
*   为什么不在读数据之前判断一下应用层缓冲区是否可以容纳内核缓冲区的全部数据
*      1.采用这种方式就会调用一次recv，传入MSG_PEEK,即recv(sockfd,, extrabuf, sizeof(extrabuf), MSG_PEEK)
*        可根据返回值判断缓冲区还有多少数据没有接收，然后再调用一次recv从内核冲读取数据
*      2.但是这样会执行两次系统调用，得不偿失，尽量使用一次系统调用就将所有数据读出，这就需要一个很大的空间
*
*   struct iovec
*      1.iov_base，存放数据的缓冲区起始位置，写时往这个位置写入iov_len个字节，读时从这个位置读出iov_len个字节
*      2.iov_len，要读入多少数据从内核缓冲区/要写入多少数据到内核缓冲区
*
*   readv(int fd, const struct iovec *iov, int iovcnt);分散读
*   writev(int fd, const struct iovec *iov, int iovcnt);集中写
*/
ssize_t Buffer::readFd(int fd, int* savedErrno)
{
// saved an ioctl()/FIONREAD call to tell how much to read
/* 开辟的栈空间，128k */
char extrabuf[65536];
/* readv用到的数据结构，定义如上 */
struct iovec vec[2];
/* 缓冲区接口，返回缓冲区还可以写入多少字节 */
const size_t writable = writableBytes();
/* 定义两块内存，一块是读缓冲区，一块是栈空间 */
vec[0].iov_base = begin()+writerIndex_;
vec[0].iov_len = writable;
vec[1].iov_base = extrabuf;
vec[1].iov_len = sizeof extrabuf;
// when there is enough space in this buffer, don't read into extrabuf.
// when extrabuf is used, we read 128k-1 bytes at most.
/* 如果应用层读缓冲区足够大(大于128k，初始时才1k -.-)，就不需要往栈区写数据了 */
const int iovcnt = (writable < sizeof extrabuf) ? 2 : 1;
/* 分散读，返回读取的字节数 */
const ssize_t n = sockets::readv(fd, vec, iovcnt);
if (n < 0)
{
*savedErrno = errno;
}
/*
* 读取的字节数比较少，读缓冲区足以容纳
* 因为读缓冲区是readv的第一块内存，所以率先向这块内存写数据
*/
else if (implicit_cast<size_t>(n) <= writable)
{
writerIndex_ += n;
}
else
{
/*
* 将栈空间的数据追加到缓冲区末尾
* 因为读缓冲区已经写满了，所以writerIndex指针就指向缓冲区的末尾
*/
writerIndex_ = buffer_.size();
append(extrabuf, n - writable);
}
// if (n == writable + sizeof extrabuf)
// {
//   goto line_30;
// }
return n;
}

append

void append(const char* /*restrict*/ data, size_t len)
{
/* 确保有足够的空间容纳len大小的数据 */
ensureWritableBytes(len);
/* 将数据copy到writerIndex后面，beginWrite返回的就是writerIndex位置的地址（writerIndex是下标） */
std::copy(data, data+len, beginWrite());
/* 写完数据，更新writerIndex */
hasWritten(len);
}

ensureWritableBytes

resize

std::vector

capacity

size

resize

capacity

void ensureWritableBytes(size_t len)
{
/* 返回剩余可用空间大小，如果不足len，开辟新空间（调用resize） */
if (writableBytes() < len)
{
makeSpace(len);
}
assert(writableBytes() >= len);
}

void makeSpace(size_t len)
{
/*
* 在多次从缓冲区读数据后，readerIndex会后移很多，导致预留空间变大
* 在增大空间之前，先判断调整预留空间的大小后能否容纳要求的数据
* 如果可以，则将预留空间缩小为8字节（默认的预留空间大小）
* 如果不可以，那么就只能增加空间
*/
if (writableBytes() + prependableBytes() < len + kCheapPrepend)
{
// FIXME: move readable data
/* writerIndex代表当前缓冲区已使用的大小，调整只需调整到恰好满足len大小即可 */
buffer_.resize(writerIndex_+len);
}
else
{
/* 通过缩小预留空间大小可以容纳len个数据，就缩小预留空间 */
// move readable data to the front, make space inside buffer
assert(kCheapPrepend < readerIndex_);
/* 返回缓冲区数据个数，writerIndex - readerIndex */
size_t readable = readableBytes();
/* 将所有数据前移 */
std::copy(begin()+readerIndex_,
begin()+writerIndex_,
begin()+kCheapPrepend);
/* 更新两个指针（下标） */
readerIndex_ = kCheapPrepend;
writerIndex_ = readerIndex_ + readable;
assert(readable == readableBytes());
}
}

readFd

Buffer

TcpServer

TcpServer

TcpConnection

TcpConnection

readFd

muduo

buf->readableBytes()

conn->name()

TcpConnection

TcpServer

receiveTime

poll

Channel

TcpConnection

onMessage

buf->retrieveAsString()

/* 从缓冲区中读取所有数据 */
string retrieveAllAsString()
{
return retrieveAsString(readableBytes());
}

/* 从缓冲区中读取len个字节的数据 */
string retrieveAsString(size_t len)
{
assert(len <= readableBytes());
/* peek返回数据的起点 */
/* 调用string(const char* s, size_type n);构造函数，初始化为从地址s开始的n个字节 */
string result(peek(), len);
/* 调整缓冲区，即改变readerIndex的位置，后移len */
retrieve(len);
return result;
}

readerIndex

/* 调整readerIndex，后移len */
void retrieve(size_t len)
{
assert(len <= readableBytes());
/*
* 如果调整后仍然有数据，就将readerIndex增加len
* 如果已经将数据全部读完(len >= readableBytes)，那么就初始化readerIndex/writerIndex位置
*/
if (len < readableBytes())
{
readerIndex_ += len;
}
else
{
retrieveAll();
}
}

readerIndex/writerIndex

/* 初始化readerIndex/writerIndex位置，通常在用户将数据全部读出之后执行 */
void retrieveAll()
{
readerIndex_ = kCheapPrepend;
writerIndex_ = kCheapPrepend;
}

TcpConnection的写操作

tcp

tcp

tcp

tcp

poll

busy loop

muduo

tcp

tcp

tcp

TcpConnection

Buffer

/* 几个重载的send函数，用于用户想要发送数据到对端 */
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)
{
/*
* 如果当前线程和TcpConnection所属线程相同，直接在当前线程发送
* 否则，需要使用std::bind绑定函数和对象，并添加到自己所在线程的事件循环中
*/
if (loop_->isInLoopThread())
{
sendInLoop(message);
}
else
{
/* 可以直接在bind中绑定函数 ? */
void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
loop_->runInLoop(
std::bind(fp,
this,     // FIXME
message.as_string()));
//std::forward<string>(message)));
}
}
}

send

sendInLoop

TcpConnection

tcp

tcp

tcp

tcp

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

/*
* 写入数据
* 1.如果Channel没有监听可写事件且输出缓冲区为空，说明之前没有出现内核缓冲区满的情况，直接写进内核
* 2.如果写入内核出错，且出错信息(errno)是EWOULDBLOCK，说明内核缓冲区满，将剩余部分添加到应用层输出缓冲区
* 3.如果之前输出缓冲区为空，那么就没有监听内核缓冲区(fd)可写事件，开始监听
*/
void TcpConnection::sendInLoop(const void* data, size_t len)
{
loop_->assertInLoopThread();
/* 写入tcp缓冲区的字节数 */
ssize_t nwrote = 0;
/* 没有写入tcp缓冲区的字节数 */
size_t remaining = len;
/* 调用write时是否出错 */
bool faultError = false;
/* 当前TcpConnection状态，TcpConnection有四种状态，kDisconnected表示已经断开连接，不能再写了，直接返回 */
if (state_ == kDisconnected)
{
LOG_WARN << "disconnected, give up writing";
return;
}
// if no thing in output queue, try writing directly
/* 如果输出缓冲区有数据，就不能尝试发送数据了，否则数据会乱，应该直接写到缓冲区中 */
if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
{
/* 读取函数 */
nwrote = sockets::write(channel_->fd(), data, len);
if (nwrote >= 0)
{
/* 写入了一些数据 */
remaining = len - nwrote;
/*
* 完全写入tcp缓冲区，且用户有提供写数据的回调函数，等待执行完后调用
* 因为当前TcpConnection和EventLoop所在同一个线程，
* 而且此时EventLoop通常处在正在处理激活Channel的过程中（当前函数有可能也是在这个过程）
* 所以等待这个函数执行完再调用回调函数
*/
if (remaining == 0 && writeCompleteCallback_)
{
loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
}
}
else // nwrote < 0
{
/* 一点也没写进去
* 如果错误为EWOULDBLOCK，表明tcp缓冲区已满
*/
nwrote = 0;
if (errno != EWOULDBLOCK)
{
/* EPIPE表示客户端已经关闭了连接，服务器仍然尝试写入，就会出现EPIPE */
LOG_SYSERR << "TcpConnection::sendInLoop";
if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
{
faultError = true;
}
}
}
}

assert(remaining <= len);
/* 没出错，且仍有一些数据没有写到tcp缓冲区中，那么就添加到写缓冲区中 */
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())
{
channel_->enableWriting();
}
}
}

tcp

tcp

Channel

TcpConnection

Channel

channel_->setWriteCallback(
std::bind(&TcpConnection::handleWrite, this));

/* 当tcp缓冲区可写时调用 */
void TcpConnection::handleWrite()
{
loop_->assertInLoopThread();
if (channel_->isWriting())
{
/* 尝试写入写缓冲区的所有数据，返回实际写入的字节数（tcp缓冲区很有可能仍然不能容纳所有数据） */
ssize_t n = sockets::write(channel_->fd(),
outputBuffer_.peek(),
outputBuffer_.readableBytes());
if (n > 0)
{
/* 调整写缓冲区的readerIndex */
outputBuffer_.retrieve(n);
if (outputBuffer_.readableBytes() == 0)
{
/* 全部写到tcp缓冲区中，关闭对可写事件的监听 */
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";
}
}

tcp

close

shutdown

close

handleWrite

shutdownInLoop

void TcpConnection::shutdownInLoop()
{
loop_->assertInLoopThread();
if (!channel_->isWriting())
{
// we are not writing
socket_->shutdownWrite();
}
}