絡(luò)緩沖區(qū)設(shè)計(jì)：雙游標(biāo)與零拷貝實(shí)現(xiàn)原理)
1. 項(xiàng)目概述為什么我們需要一個(gè)專門的網(wǎng)絡(luò)緩沖區(qū)做網(wǎng)絡(luò)編程尤其是用C寫高性能服務(wù)器你遲早會(huì)碰到一個(gè)繞不開的核心組件網(wǎng)絡(luò)緩沖區(qū)Network Buffer。這東西聽起來簡(jiǎn)單不就是一塊內(nèi)存用來存數(shù)據(jù)嗎但真要自己動(dòng)手設(shè)計(jì)一個(gè)你會(huì)發(fā)現(xiàn)坑多得能絆倒一頭大象。我見過太多新手寫的服務(wù)器要么是每個(gè)連接上來就new char[1024]要么是直接用std::string來收發(fā)包結(jié)果在高并發(fā)下性能慘不忍睹內(nèi)存碎片化嚴(yán)重甚至直接崩潰。網(wǎng)絡(luò)緩沖區(qū)的本質(zhì)是為應(yīng)用層和操作系統(tǒng)內(nèi)核的TCP/IP協(xié)議棧之間建立一個(gè)高效、可控的數(shù)據(jù)中轉(zhuǎn)站。網(wǎng)絡(luò)I/O特別是套接字Socket讀寫其核心矛盾是“生產(chǎn)者”網(wǎng)絡(luò)對(duì)端和“消費(fèi)者”我們的業(yè)務(wù)邏輯速度不匹配以及操作系統(tǒng)read/write或recv/send調(diào)用的離散性與業(yè)務(wù)邏輯需要處理完整消息之間的矛盾。一個(gè)設(shè)計(jì)良好的緩沖區(qū)能平滑這種速度差將零碎的數(shù)據(jù)包組裝成完整的應(yīng)用層報(bào)文同時(shí)管理好內(nèi)存的生命周期避免頻繁的系統(tǒng)調(diào)用和內(nèi)存分配。簡(jiǎn)單來說它要解決幾個(gè)核心痛點(diǎn)1)減少系統(tǒng)調(diào)用攢夠數(shù)據(jù)再處理避免為了一兩個(gè)字節(jié)就調(diào)用read。2)方便消息解析提供游標(biāo)如readIndex/writeIndex方便地切割出完整報(bào)文。3)高效內(nèi)存管理避免小內(nèi)存頻繁申請(qǐng)釋放通常采用預(yù)分配動(dòng)態(tài)擴(kuò)容的機(jī)制。4)線程安全可選為可能的跨線程數(shù)據(jù)交換做準(zhǔn)備。在C的世界里實(shí)現(xiàn)這樣一個(gè)緩沖區(qū)是對(duì)你內(nèi)存管理、數(shù)據(jù)結(jié)構(gòu)和API設(shè)計(jì)能力的綜合考驗(yàn)。下面我就結(jié)合自己踩過的坑拆解一個(gè)工業(yè)級(jí)網(wǎng)絡(luò)緩沖區(qū)的設(shè)計(jì)思路與C實(shí)現(xiàn)細(xì)節(jié)。2. 核心設(shè)計(jì)思路與數(shù)據(jù)結(jié)構(gòu)選型設(shè)計(jì)一個(gè)緩沖區(qū)首先得想清楚它長(zhǎng)什么樣、怎么用。一個(gè)常見的誤區(qū)是直接拿std::vectorchar當(dāng)緩沖區(qū)。vector的連續(xù)內(nèi)存和動(dòng)態(tài)擴(kuò)容看起來很美但在網(wǎng)絡(luò)編程場(chǎng)景下有致命缺陷當(dāng)你從頭部pop掉已處理的數(shù)據(jù)后騰出的空間無法被后續(xù)寫入直接利用除非你頻繁地erase頭部元素這會(huì)導(dǎo)致大量?jī)?nèi)存拷貝。我們需要的是一個(gè)更接近“循環(huán)隊(duì)列”或“雙端隊(duì)列”概念的結(jié)構(gòu)但實(shí)現(xiàn)上要更高效。2.1 主流設(shè)計(jì)模式預(yù)分配連續(xù)內(nèi)存與雙游標(biāo)經(jīng)過多年實(shí)踐業(yè)界如muduo庫、Netty形成了一種經(jīng)典且高效的緩沖區(qū)設(shè)計(jì)模式內(nèi)部維護(hù)一塊連續(xù)的預(yù)分配內(nèi)存如std::vectorchar或char[]用兩個(gè)游標(biāo)索引來標(biāo)記可讀數(shù)據(jù)的起始位置readIndex和可寫空間的起始位置writeIndex。----------------------------------------------------------------------- | 已讀/可回收區(qū)域 | 可讀數(shù)據(jù) (readable) | 可寫空間 (writable) | ----------------------------------------------------------------------- ^ ^ ^ 0 readIndex writeIndex capacityreadIndex: 指向緩沖區(qū)中下一個(gè)待讀取已接收但尚未被應(yīng)用層消費(fèi)的字節(jié)。writeIndex: 指向緩沖區(qū)中下一個(gè)可寫入空閑字節(jié)的位置。capacity: 緩沖區(qū)的總?cè)萘?。這樣設(shè)計(jì)的好處是內(nèi)存連續(xù)便于直接傳遞給系統(tǒng)調(diào)用如readv,write或進(jìn)行內(nèi)存操作如memcpy。零拷貝應(yīng)用層解析報(bào)文時(shí)可以直接基于readIndex指針操作無需將數(shù)據(jù)拷貝出來。高效空間復(fù)用當(dāng)readIndex前進(jìn)消費(fèi)了數(shù)據(jù)它前面的空間就變成了“可回收區(qū)域”。當(dāng)可寫空間不足時(shí)我們可以嘗試將可讀數(shù)據(jù)移動(dòng)到緩沖區(qū)頭部memmove從而在不重新分配內(nèi)存的情況下騰出尾部空間。這比vector的頭部刪除高效得多。邏輯清晰readableBytes() writeIndex - readIndex,writableBytes() capacity - writeIndex狀態(tài)一目了然。2.2 數(shù)據(jù)結(jié)構(gòu)選型為什么不用std::deque有人會(huì)問std::deque不也是雙端操作高效嗎為什么不用它這涉及到網(wǎng)絡(luò)編程的另一個(gè)核心需求與系統(tǒng)I/O接口的直接交互。系統(tǒng)調(diào)用如read( fd, buf, len )或更高效的readv需要的是連續(xù)的內(nèi)存地址。deque的內(nèi)部結(jié)構(gòu)是分段連續(xù)的無法直接提供一個(gè)大的、連續(xù)的內(nèi)存塊給readv一次寫入。雖然可以通過gather I/Owritev處理不連續(xù)輸出但對(duì)于輸入我們更希望數(shù)據(jù)能連續(xù)地落在一塊內(nèi)存里方便后續(xù)解析。因此自己管理一塊連續(xù)內(nèi)存在性能和可控性上是最好的選擇。2.3 內(nèi)存管理策略預(yù)分配與動(dòng)態(tài)擴(kuò)容緩沖區(qū)的初始大小kInitialSize是個(gè)經(jīng)驗(yàn)值比如1024或4096字節(jié)。太小會(huì)導(dǎo)致頻繁擴(kuò)容太大可能浪費(fèi)內(nèi)存。擴(kuò)容策略是關(guān)鍵何時(shí)擴(kuò)容當(dāng)writableBytes()小于本次期望寫入的數(shù)據(jù)量時(shí)。如何擴(kuò)容不是簡(jiǎn)單realloc。我們的策略是先檢查頭部可回收區(qū)域尾部可寫空間的總和是否夠用。如果夠就執(zhí)行一次memmove將可讀數(shù)據(jù)移動(dòng)到緩沖區(qū)頭部騰出尾部空間。這稱為內(nèi)部騰挪。只有當(dāng)內(nèi)部騰挪后空間仍不足時(shí)才真正分配一塊更大的新內(nèi)存比如翻倍拷貝數(shù)據(jù)過去釋放舊內(nèi)存。避免抖動(dòng)擴(kuò)容因子如2倍要合適避免微小數(shù)據(jù)增長(zhǎng)導(dǎo)致頻繁擴(kuò)容。3. 核心接口設(shè)計(jì)與C實(shí)現(xiàn)細(xì)節(jié)有了設(shè)計(jì)思路我們來看具體實(shí)現(xiàn)。我將一個(gè)緩沖區(qū)類命名為Buffer它應(yīng)該提供以下幾組核心接口。3.1 構(gòu)造函數(shù)與內(nèi)存初始化class Buffer { public: static const size_t kCheapPrepend 8; // 預(yù)留空間可用于存消息長(zhǎng)度等 static const size_t kInitialSize 1024; // 初始可寫緩沖區(qū)大小 explicit Buffer(size_t initialSize kInitialSize) : buffer_(kCheapPrepend initialSize), // 總?cè)萘? 預(yù)留 初始 readerIndex_(kCheapPrepend), // 讀索引從預(yù)留后開始 writerIndex_(kCheapPrepend) { // 寫索引同上 assert(readableBytes() 0); assert(writableBytes() initialSize); assert(prependableBytes() kCheapPrepend); } // ... 其他成員函數(shù) private: std::vectorchar buffer_; // 底層存儲(chǔ) size_t readerIndex_; size_t writerIndex_; // 靜態(tài)斷言確保索引類型足夠大 static_assert(sizeof(readerIndex_) sizeof(size_t), 索引類型太小); };要點(diǎn)解析kCheapPrepend這是一個(gè)非常巧妙的設(shè)計(jì)。它在緩沖區(qū)最前面預(yù)留了一小段空間比如8字節(jié)。為什么在有些網(wǎng)絡(luò)協(xié)議中如自定義協(xié)議我們希望在應(yīng)用層報(bào)文前面加一個(gè)長(zhǎng)度字段。有了這個(gè)預(yù)留空間我們?cè)诮M裝消息時(shí)可以先把報(bào)文體append到緩沖區(qū)最后再計(jì)算長(zhǎng)度并prepend到預(yù)留空間里整個(gè)過程不需要移動(dòng)數(shù)據(jù)效率極高。使用std::vectorchar作為底層容器利用其RAII特性自動(dòng)管理內(nèi)存生命周期。初始狀態(tài)可讀數(shù)據(jù)為0可寫空間為initialSize可預(yù)留空間為kCheapPrepend。3.2 基礎(chǔ)狀態(tài)查詢接口這些接口是緩沖區(qū)運(yùn)轉(zhuǎn)的基礎(chǔ)必須高效inline且無誤。size_t readableBytes() const { return writerIndex_ - readerIndex_; } size_t writableBytes() const { return buffer_.size() - writerIndex_; } size_t prependableBytes() const { return readerIndex_; } // 頭部可回收/預(yù)留空間 const char* peek() const { return begin() readerIndex_; } // 獲取可讀數(shù)據(jù)首指針 char* beginWrite() { return begin() writerIndex_; } const char* beginWrite() const { return begin() writerIndex_; }begin()輔助函數(shù)返回buffer_的底層指針return *buffer_.begin();注意vector在為空時(shí)begin()可能未定義但我們構(gòu)造函數(shù)已分配內(nèi)存所以安全。3.3 核心操作retrieve,append,prepend這是緩沖區(qū)數(shù)據(jù)流動(dòng)的三大操作。1.retrieve消費(fèi)數(shù)據(jù)當(dāng)應(yīng)用層從緩沖區(qū)取走解析完len字節(jié)數(shù)據(jù)后需要移動(dòng)readerIndex_。void retrieve(size_t len) { if (len readableBytes()) { readerIndex_ len; // 消費(fèi)部分?jǐn)?shù)據(jù) } else { // len readableBytes() retrieveAll(); } } void retrieveAll() { readerIndex_ kCheapPrepend; writerIndex_ kCheapPrepend; }關(guān)鍵點(diǎn)retrieve并不真的刪除或清零數(shù)據(jù)只是移動(dòng)索引。這符合“零拷貝”思想。retrieveAll()非常高效直接將索引復(fù)位到初始狀態(tài)準(zhǔn)備接收新數(shù)據(jù)。2.append寫入數(shù)據(jù)這是最常用的操作將數(shù)據(jù)從外部寫入緩沖區(qū)的可寫區(qū)域。void append(const char* data, size_t len) { ensureWritableBytes(len); // 確?？臻g足夠可能觸發(fā)擴(kuò)容或騰挪 std::copy(data, data len, beginWrite()); hasWritten(len); // 移動(dòng)writerIndex_ } void ensureWritableBytes(size_t len) { if (writableBytes() len) { makeSpace(len); // 核心騰挪或擴(kuò)容 } } void hasWritten(size_t len) { writerIndex_ len; }makeSpace的實(shí)現(xiàn)是精華所在void makeSpace(size_t len) { // 情況1頭部預(yù)留空間尾部可寫空間足夠 if (writableBytes() prependableBytes() len kCheapPrepend) { // 不夠需要重新分配 buffer_.resize(writerIndex_ len); } else { // 夠進(jìn)行內(nèi)部騰挪 size_t readable readableBytes(); std::copy(begin() readerIndex_, begin() writerIndex_, begin() kCheapPrepend); readerIndex_ kCheapPrepend; writerIndex_ readerIndex_ readable; assert(readable readableBytes()); } }3.prepend向前寫入數(shù)據(jù)利用預(yù)留空間在可讀數(shù)據(jù)前面插入數(shù)據(jù)常用于添加協(xié)議頭。void prepend(const void* data, size_t len) { assert(len prependableBytes()); // 必須保證有足夠預(yù)留空間 readerIndex_ - len; const char* d static_castconst char*(data); std::copy(d, d len, begin() readerIndex_); }3.4 與網(wǎng)絡(luò)I/O的對(duì)接readFd與writeFd這是緩沖區(qū)價(jià)值體現(xiàn)的關(guān)鍵它封裝了與Socket文件描述符fd的交互。從Socket讀取數(shù)據(jù)readFd一個(gè)健壯的readFd需要處理TCP粘包、內(nèi)核緩沖區(qū)大小、以及EAGAIN/EWOULDBLOCK等問題。這里展示一個(gè)使用棧上額外緩沖區(qū)和readv的經(jīng)典高效實(shí)現(xiàn)// 返回讀取的字節(jié)數(shù)-1表示錯(cuò)誤errno需檢查0表示對(duì)端關(guān)閉 ssize_t Buffer::readFd(int fd, int* savedErrno) { // 棧上準(zhǔn)備一個(gè)額外緩沖區(qū)用于應(yīng)對(duì)緩沖區(qū)不夠的情況 char extrabuf[65536]; // 64K struct iovec vec[2]; const size_t writable writableBytes(); // 第一塊內(nèi)存Buffer自身的可寫空間 vec[0].iov_base beginWrite(); vec[0].iov_len writable; // 第二塊內(nèi)存棧上緩沖區(qū) vec[1].iov_base extrabuf; vec[1].iov_len sizeof(extrabuf); // 當(dāng)Buffer可寫空間較小時(shí)使用readv分散讀 const int iovcnt (writable sizeof(extrabuf)) ? 2 : 1; const ssize_t n ::readv(fd, vec, iovcnt); if (n 0) { *savedErrno errno; } else if (n writable) { // 數(shù)據(jù)全部讀到了Buffer的可寫空間 writerIndex_ n; } else { // 數(shù)據(jù)填滿了Buffer的可寫空間并有一部分讀到了extrabuf writerIndex_ buffer_.size(); // Buffer寫滿 append(extrabuf, n - writable); // 將extrabuf中的數(shù)據(jù)追加進(jìn)來會(huì)觸發(fā)擴(kuò)容 } return n; }為什么用readv和棧上緩沖區(qū)一次系統(tǒng)調(diào)用讀更多數(shù)據(jù)readv允許我們將數(shù)據(jù)分散讀到多個(gè)內(nèi)存塊。如果Buffer自身的空間足夠就只讀到Buffer里如果不夠剩余的數(shù)據(jù)可以讀到棧上的extrabuf避免因?yàn)锽uffer一時(shí)空間不足而丟失數(shù)據(jù)或觸發(fā)擴(kuò)容擴(kuò)容是堆操作更慢。棧上緩沖區(qū)速度快extrabuf在棧上分配和釋放極快。即使數(shù)據(jù)先到了extrabuf隨后再append到Buffer也多了一次內(nèi)存拷貝但這比因?yàn)锽uffer空間不足導(dǎo)致內(nèi)核數(shù)據(jù)無法及時(shí)讀出、觸發(fā)下一次EPOLLIN事件水平觸發(fā)模式下或等待邊緣觸發(fā)模式下要高效得多。這是一種用空間棧內(nèi)存換時(shí)間處理效率和編程復(fù)雜度的權(quán)衡。處理粘包readFd只負(fù)責(zé)把內(nèi)核緩沖區(qū)數(shù)據(jù)盡可能多地讀到應(yīng)用層緩沖區(qū)不負(fù)責(zé)解析消息邊界。消息邊界的解析如根據(jù)長(zhǎng)度字段或分隔符應(yīng)由應(yīng)用層在readFd之后通過peek()和retrieve()來完成。向Socket寫入數(shù)據(jù)writeFd將Buffer中可讀的數(shù)據(jù)通過Socket發(fā)送出去。// 返回寫入的字節(jié)數(shù)-1表示錯(cuò)誤需檢查errno ssize_t Buffer::writeFd(int fd, int* savedErrno) { size_t nLeft readableBytes(); ssize_t nWritten 0; const char* bufPtr peek(); while (nLeft 0) { nWritten ::write(fd, bufPtr, nLeft); if (nWritten 0) { if (errno EINTR) { // 被信號(hào)中斷重試 continue; } else if (errno EAGAIN || errno EWOULDBLOCK) { // 非阻塞模式下寫緩沖區(qū)已滿 break; } else { *savedErrno errno; // 其他錯(cuò)誤 return -1; } } nLeft - nWritten; bufPtr nWritten; retrieve(nWritten); // 重要發(fā)送成功的數(shù)據(jù)要從Buffer中消費(fèi)掉 } return (readableBytes() 0) ? 0 : (nWritten 0 ? nWritten : -1); // 返回0表示所有數(shù)據(jù)已寫完或EAGAIN-1表示錯(cuò)誤0表示部分寫入在非阻塞模式下可能發(fā)生 }關(guān)鍵點(diǎn)循環(huán)寫入因?yàn)閣rite系統(tǒng)調(diào)用可能只寫入部分?jǐn)?shù)據(jù)尤其是非阻塞Socket所以需要循環(huán)直到所有數(shù)據(jù)寫完或遇到EAGAIN。及時(shí)retrieve成功寫入多少字節(jié)就要從Buffer中消費(fèi)掉多少字節(jié)。否則下次writeFd會(huì)重復(fù)發(fā)送已發(fā)送的數(shù)據(jù)。錯(cuò)誤處理區(qū)分EINTR系統(tǒng)調(diào)用被信號(hào)中斷可重試、EAGAIN/EWOULDBLOCK非阻塞Socket寫緩沖區(qū)滿是正常情況應(yīng)停止循環(huán)等待下次可寫事件和其他致命錯(cuò)誤。4. 高級(jí)特性與性能優(yōu)化考量一個(gè)基礎(chǔ)的Buffer類已經(jīng)能工作但要用于生產(chǎn)環(huán)境還需要考慮更多。4.1 移動(dòng)語義與零拷貝優(yōu)化現(xiàn)代C強(qiáng)調(diào)移動(dòng)語義。我們的Buffer管理著vector應(yīng)該實(shí)現(xiàn)移動(dòng)構(gòu)造函數(shù)和移動(dòng)賦值運(yùn)算符避免不必要的深拷貝。Buffer(Buffer other) noexcept : buffer_(std::move(other.buffer_)), readerIndex_(other.readerIndex_), writerIndex_(other.writerIndex_) { other.readerIndex_ other.writerIndex_ kCheapPrepend; // 將other置于有效但空的狀態(tài) } Buffer operator(Buffer rhs) noexcept { if (this ! rhs) { buffer_ std::move(rhs.buffer_); readerIndex_ rhs.readerIndex_; writerIndex_ rhs.writerIndex_; rhs.readerIndex_ rhs.writerIndex_ kCheapPrepend; } return *this; } // 禁用拷貝構(gòu)造和拷貝賦值因?yàn)榭截惓杀靖咄ǔ２恍枰?Buffer(const Buffer) delete; Buffer operator(const Buffer) delete;對(duì)于某些極端性能場(chǎng)景可以考慮更激進(jìn)的零拷貝。例如如果有一個(gè)超大消息要發(fā)送可以不append到Buffer而是直接用一個(gè)std::vectorchar或std::string持有并記錄其生命周期在writeFd時(shí)使用writev將其與Buffer中的數(shù)據(jù)一并發(fā)送。但這會(huì)大大增加接口復(fù)雜性和生命周期管理的難度除非有確切的性能瓶頸否則不建議在基礎(chǔ)Buffer類中實(shí)現(xiàn)。4.2 線程安全性默認(rèn)情況下這個(gè)Buffer不是線程安全的。它被設(shè)計(jì)為每個(gè)TCP連接獨(dú)占一個(gè)Buffer實(shí)例而一個(gè)連接通常只在一個(gè)IO線程中被處理Reactor模式因此不需要內(nèi)部加鎖。如果你需要在多個(gè)線程間傳遞Buffer那么應(yīng)該由外層例如使用它的網(wǎng)絡(luò)庫或應(yīng)用通過消息隊(duì)列等同步機(jī)制來保證而不是在Buffer內(nèi)部加鎖。內(nèi)部加鎖會(huì)嚴(yán)重?fù)p害性能且鎖粒度難以控制。4.3 內(nèi)存池集成在高并發(fā)場(chǎng)景下即使有內(nèi)部騰挪頻繁的vector::resize涉及new/delete也可能成為瓶頸。一個(gè)優(yōu)化方向是集成內(nèi)存池Memory Pool。例如可以預(yù)分配一大塊內(nèi)存char數(shù)組然后以鏈表或數(shù)組管理多個(gè)固定大小的Buffer塊。當(dāng)Buffer需要擴(kuò)容時(shí)不是重新分配而是從內(nèi)存池中再申請(qǐng)一個(gè)塊并以鏈表形式連接起來這類似于std::deque但塊的大小是固定的、池化的。這能顯著減少系統(tǒng)調(diào)用malloc/free的次數(shù)但代價(jià)是Buffer不再是連續(xù)內(nèi)存peek()和beginWrite()可能失效需要更復(fù)雜的迭代器設(shè)計(jì)。這屬于高級(jí)優(yōu)化需要根據(jù)實(shí)際性能分析來決定是否引入。4.4 序列化與反序列化輔助Buffer可以作為簡(jiǎn)單的序列化工具。我們可以為其添加針對(duì)基本類型的append和retrieve重載。void appendInt32(int32_t x) { int32_t be32 htobe32(x); // 轉(zhuǎn)換為網(wǎng)絡(luò)字節(jié)序大端 append(be32, sizeof(be32)); } int32_t readInt32() { int32_t result peekInt32(); retrieve(sizeof(int32_t)); return result; } int32_t peekInt32() const { assert(readableBytes() sizeof(int32_t)); int32_t be32 0; ::memcpy(be32, peek(), sizeof(be32)); return be32toh(be32); // 轉(zhuǎn)換為主機(jī)字節(jié)序 }這樣應(yīng)用層協(xié)議編解碼會(huì)方便很多。注意字節(jié)序轉(zhuǎn)換htonl,ntohl等是網(wǎng)絡(luò)編程的必備步驟。5. 實(shí)戰(zhàn)應(yīng)用構(gòu)建一個(gè)簡(jiǎn)單的回顯服務(wù)器理論說再多不如看實(shí)際怎么用。我們用一個(gè)最簡(jiǎn)單的TCP回顯服務(wù)器Echo Server來演示Buffer的完整工作流程。這個(gè)服務(wù)器使用非阻塞IO和epollLinux或kqueueBSD的事件驅(qū)動(dòng)模型每個(gè)連接使用一個(gè)Buffer。核心事件循環(huán)偽代碼void onConnection(const TcpConnectionPtr conn) { if (conn-connected()) { // 為新連接分配一個(gè)Buffer conn-setContext(Buffer()); // 上下文綁定一個(gè)Buffer實(shí)例 } } void onMessage(const TcpConnectionPtr conn, Buffer* buf, Timestamp) { // buf 就是該連接對(duì)應(yīng)的Buffer // 1. 嘗試從buf中解析一條完整消息這里假設(shè)是換行符分隔 const char* crlf buf-findCRLF(); // 我們需要在Buffer里實(shí)現(xiàn)一個(gè)findCRLF方法 if (crlf) { // 2. 找到一條消息 size_t len crlf - buf-peek(); std::string message(buf-peek(), len); // 拷貝出來實(shí)際可零拷貝處理 buf-retrieve(len 2); // 消費(fèi)掉消息包括\r\n // 3. 處理消息這里簡(jiǎn)單回顯 LOG_INFO Echo message.size() bytes; conn-send(message \r\n); // conn-send內(nèi)部會(huì)調(diào)用我們Buffer的append } // 4. 如果buf中還有數(shù)據(jù)粘包下次onMessage會(huì)繼續(xù)處理 } void onWriteComplete(const TcpConnectionPtr conn) { // 數(shù)據(jù)發(fā)送完畢后的回調(diào)可用于流量控制等 }findCRLF的實(shí)現(xiàn)示例const char* Buffer::findCRLF() const { const char* crlf std::search(peek(), beginWrite(), kCRLF, kCRLF2); return crlf beginWrite() ? nullptr : crlf; } // 同理可以實(shí)現(xiàn)findEOL等在這個(gè)流程中onMessage被調(diào)用時(shí)連接對(duì)應(yīng)的Buffer里已經(jīng)由網(wǎng)絡(luò)庫通過readFd填入了最新的數(shù)據(jù)。我們嘗試從Buffer中查找消息邊界這里是\r\n。找到后取出消息內(nèi)容進(jìn)行處理回顯并調(diào)用retrieve消費(fèi)掉這部分?jǐn)?shù)據(jù)。如果沒找到完整消息說明數(shù)據(jù)還沒收全什么也不做等待下一次onMessage。發(fā)送數(shù)據(jù)時(shí)conn-send會(huì)將數(shù)據(jù)append到該連接的輸出緩沖區(qū)另一個(gè)Buffer實(shí)例然后網(wǎng)絡(luò)庫會(huì)在Socket可寫時(shí)調(diào)用該輸出緩沖區(qū)的writeFd將數(shù)據(jù)發(fā)送出去。這就是Buffer在網(wǎng)絡(luò)編程中的典型作用作為輸入輸出的數(shù)據(jù)暫存地和消息邊界解析器。6. 常見陷阱、調(diào)試技巧與性能測(cè)試即使設(shè)計(jì)再完善實(shí)際使用中還是會(huì)遇到各種問題。這里分享幾個(gè)我踩過的坑和解決方法。6.1 典型陷阱索引越界這是最常出現(xiàn)的bug。任何對(duì)readerIndex_和writerIndex_的修改都必須前置條件檢查assert在調(diào)試期很有用。確保readerIndex_ writerIndex_ buffer_.size()始終成立。忘記retrieve在writeFd成功發(fā)送數(shù)據(jù)后一定要調(diào)用retrieve。否則Buffer會(huì)認(rèn)為那些數(shù)據(jù)還沒被消費(fèi)導(dǎo)致內(nèi)存泄漏邏輯上和重復(fù)發(fā)送。擴(kuò)容策略激進(jìn)如果擴(kuò)容因子太大比如10倍可能一次性占用過多內(nèi)存。如果太小比如每次加100字節(jié)又會(huì)頻繁擴(kuò)容。需要根據(jù)業(yè)務(wù)流量特點(diǎn)進(jìn)行壓測(cè)和調(diào)整。線程安全誤用如前所述這個(gè)Buffer不是線程安全的。如果你在A線程append數(shù)據(jù)在B線程retrieve必須在外層加鎖。readFd中extrabuf大小?？臻g有限通常幾MBextrabuf不能太大如1MB否則有棧溢出風(fēng)險(xiǎn)。64KB是一個(gè)經(jīng)驗(yàn)值平衡了性能和安全性。6.2 調(diào)試技巧添加狀態(tài)打印函數(shù)在調(diào)試時(shí)一個(gè)dump()函數(shù)非常有用。void Buffer::dump() const { printf(Buffer size: %zu, readable: %zu, writable: %zu, prependable: %zu\n, buffer_.size(), readableBytes(), writableBytes(), prependableBytes()); printf(ReaderIndex: %zu, WriterIndex: %zu\n, readerIndex_, writerIndex_); // 可選以16進(jìn)制打印可讀數(shù)據(jù) for (size_t i readerIndex_; i writerIndex_; i) { printf(%02x , static_castunsigned char(buffer_[i])); if ((i - readerIndex_ 1) % 16 0) printf(\n); } printf(\n); }使用Valgrind或AddressSanitizer檢查內(nèi)存錯(cuò)誤如越界訪問、使用未初始化內(nèi)存等。壓力測(cè)試與內(nèi)存監(jiān)控編寫測(cè)試程序模擬海量連接和不同大小的數(shù)據(jù)包收發(fā)使用top、pmap或valgrind --toolmassif監(jiān)控內(nèi)存增長(zhǎng)和碎片情況。6.3 性能測(cè)試關(guān)注點(diǎn)設(shè)計(jì)一個(gè)簡(jiǎn)單的性能測(cè)試對(duì)比不同實(shí)現(xiàn)如直接read/write、使用std::string、使用我們的Buffer的差異。吞吐量Throughput在固定時(shí)間內(nèi)能處理多少數(shù)據(jù)。延遲Latency處理一個(gè)請(qǐng)求的平均時(shí)間。內(nèi)存占用Memory Footprint在處理大量連接時(shí)每個(gè)連接Buffer的內(nèi)存開銷。CPU使用率特別是在內(nèi)部騰挪和擴(kuò)容時(shí)的CPU消耗。測(cè)試時(shí)可以模擬不同的消息大小從幾個(gè)字節(jié)到幾十KB和發(fā)送頻率突發(fā)、勻速觀察Buffer的表現(xiàn)。通常我們的Buffer實(shí)現(xiàn)在處理大量小包和突發(fā)大包時(shí)會(huì)比樸素的方案穩(wěn)定得多。最后網(wǎng)絡(luò)緩沖區(qū)的設(shè)計(jì)沒有銀彈上述實(shí)現(xiàn)是一個(gè)在通用性、性能和復(fù)雜度之間取得較好平衡的方案。在實(shí)際項(xiàng)目中你可能還需要根據(jù)具體協(xié)議如HTTP/2的幀結(jié)構(gòu)、硬件特性如DPDK用戶態(tài)網(wǎng)絡(luò)進(jìn)行特化和優(yōu)化。但萬變不離其宗理解其核心設(shè)計(jì)思想——連續(xù)內(nèi)存、雙游標(biāo)、內(nèi)部騰挪、與I/O高效對(duì)接——將幫助你在面對(duì)任何網(wǎng)絡(luò)編程任務(wù)時(shí)都能設(shè)計(jì)出合適的數(shù)據(jù)緩沖區(qū)。 C++高性能網(wǎng)絡(luò)緩沖區(qū)設(shè)計(jì)：雙游標(biāo)與零拷貝實(shí)現(xiàn)原理
1. 項(xiàng)目概述為什么我們需要一個(gè)專門的網(wǎng)絡(luò)緩沖區(qū)做網(wǎng)絡(luò)編程尤其是用C寫高性能服務(wù)器你遲早會(huì)碰到一個(gè)繞不開的核心組件網(wǎng)絡(luò)緩沖區(qū)Network Buffer。這東西聽起來簡(jiǎn)單不就是一塊內(nèi)存用來存數(shù)據(jù)嗎但真要自己動(dòng)手設(shè)計(jì)一個(gè)你會(huì)發(fā)現(xiàn)坑多得能絆倒一頭大象。我見過太多新手寫的服務(wù)器要么是每個(gè)連接上來就new char[1024]要么是直接用std::string來收發(fā)包結(jié)果在高并發(fā)下性能慘不忍睹內(nèi)存碎片化嚴(yán)重甚至直接崩潰。網(wǎng)絡(luò)緩沖區(qū)的本質(zhì)是為應(yīng)用層和操作系統(tǒng)內(nèi)核的TCP/IP協(xié)議棧之間建立一個(gè)高效、可控的數(shù)據(jù)中轉(zhuǎn)站。網(wǎng)絡(luò)I/O特別是套接字Socket讀寫其核心矛盾是“生產(chǎn)者”網(wǎng)絡(luò)對(duì)端和“消費(fèi)者”我們的業(yè)務(wù)邏輯速度不匹配以及操作系統(tǒng)read/write或recv/send調(diào)用的離散性與業(yè)務(wù)邏輯需要處理完整消息之間的矛盾。一個(gè)設(shè)計(jì)良好的緩沖區(qū)能平滑這種速度差將零碎的數(shù)據(jù)包組裝成完整的應(yīng)用層報(bào)文同時(shí)管理好內(nèi)存的生命周期避免頻繁的系統(tǒng)調(diào)用和內(nèi)存分配。簡(jiǎn)單來說它要解決幾個(gè)核心痛點(diǎn)1)減少系統(tǒng)調(diào)用攢夠數(shù)據(jù)再處理避免為了一兩個(gè)字節(jié)就調(diào)用read。2)方便消息解析提供游標(biāo)如readIndex/writeIndex方便地切割出完整報(bào)文。3)高效內(nèi)存管理避免小內(nèi)存頻繁申請(qǐng)釋放通常采用預(yù)分配動(dòng)態(tài)擴(kuò)容的機(jī)制。4)線程安全可選為可能的跨線程數(shù)據(jù)交換做準(zhǔn)備。在C的世界里實(shí)現(xiàn)這樣一個(gè)緩沖區(qū)是對(duì)你內(nèi)存管理、數(shù)據(jù)結(jié)構(gòu)和API設(shè)計(jì)能力的綜合考驗(yàn)。下面我就結(jié)合自己踩過的坑拆解一個(gè)工業(yè)級(jí)網(wǎng)絡(luò)緩沖區(qū)的設(shè)計(jì)思路與C實(shí)現(xiàn)細(xì)節(jié)。2. 核心設(shè)計(jì)思路與數(shù)據(jù)結(jié)構(gòu)選型設(shè)計(jì)一個(gè)緩沖區(qū)首先得想清楚它長(zhǎng)什么樣、怎么用。一個(gè)常見的誤區(qū)是直接拿std::vectorchar當(dāng)緩沖區(qū)。vector的連續(xù)內(nèi)存和動(dòng)態(tài)擴(kuò)容看起來很美但在網(wǎng)絡(luò)編程場(chǎng)景下有致命缺陷當(dāng)你從頭部pop掉已處理的數(shù)據(jù)后騰出的空間無法被后續(xù)寫入直接利用除非你頻繁地erase頭部元素這會(huì)導(dǎo)致大量?jī)?nèi)存拷貝。我們需要的是一個(gè)更接近“循環(huán)隊(duì)列”或“雙端隊(duì)列”概念的結(jié)構(gòu)但實(shí)現(xiàn)上要更高效。2.1 主流設(shè)計(jì)模式預(yù)分配連續(xù)內(nèi)存與雙游標(biāo)經(jīng)過多年實(shí)踐業(yè)界如muduo庫、Netty形成了一種經(jīng)典且高效的緩沖區(qū)設(shè)計(jì)模式內(nèi)部維護(hù)一塊連續(xù)的預(yù)分配內(nèi)存如std::vectorchar或char[]用兩個(gè)游標(biāo)索引來標(biāo)記可讀數(shù)據(jù)的起始位置readIndex和可寫空間的起始位置writeIndex。----------------------------------------------------------------------- | 已讀/可回收區(qū)域 | 可讀數(shù)據(jù) (readable) | 可寫空間 (writable) | ----------------------------------------------------------------------- ^ ^ ^ 0 readIndex writeIndex capacityreadIndex: 指向緩沖區(qū)中下一個(gè)待讀取已接收但尚未被應(yīng)用層消費(fèi)的字節(jié)。writeIndex: 指向緩沖區(qū)中下一個(gè)可寫入空閑字節(jié)的位置。capacity: 緩沖區(qū)的總?cè)萘?。這樣設(shè)計(jì)的好處是內(nèi)存連續(xù)便于直接傳遞給系統(tǒng)調(diào)用如readv,write或進(jìn)行內(nèi)存操作如memcpy。零拷貝應(yīng)用層解析報(bào)文時(shí)可以直接基于readIndex指針操作無需將數(shù)據(jù)拷貝出來。高效空間復(fù)用當(dāng)readIndex前進(jìn)消費(fèi)了數(shù)據(jù)它前面的空間就變成了“可回收區(qū)域”。當(dāng)可寫空間不足時(shí)我們可以嘗試將可讀數(shù)據(jù)移動(dòng)到緩沖區(qū)頭部memmove從而在不重新分配內(nèi)存的情況下騰出尾部空間。這比vector的頭部刪除高效得多。邏輯清晰readableBytes() writeIndex - readIndex,writableBytes() capacity - writeIndex狀態(tài)一目了然。2.2 數(shù)據(jù)結(jié)構(gòu)選型為什么不用std::deque有人會(huì)問std::deque不也是雙端操作高效嗎為什么不用它這涉及到網(wǎng)絡(luò)編程的另一個(gè)核心需求與系統(tǒng)I/O接口的直接交互。系統(tǒng)調(diào)用如read( fd, buf, len )或更高效的readv需要的是連續(xù)的內(nèi)存地址。deque的內(nèi)部結(jié)構(gòu)是分段連續(xù)的無法直接提供一個(gè)大的、連續(xù)的內(nèi)存塊給readv一次寫入。雖然可以通過gather I/Owritev處理不連續(xù)輸出但對(duì)于輸入我們更希望數(shù)據(jù)能連續(xù)地落在一塊內(nèi)存里方便后續(xù)解析。因此自己管理一塊連續(xù)內(nèi)存在性能和可控性上是最好的選擇。2.3 內(nèi)存管理策略預(yù)分配與動(dòng)態(tài)擴(kuò)容緩沖區(qū)的初始大小kInitialSize是個(gè)經(jīng)驗(yàn)值比如1024或4096字節(jié)。太小會(huì)導(dǎo)致頻繁擴(kuò)容太大可能浪費(fèi)內(nèi)存。擴(kuò)容策略是關(guān)鍵何時(shí)擴(kuò)容當(dāng)writableBytes()小于本次期望寫入的數(shù)據(jù)量時(shí)。如何擴(kuò)容不是簡(jiǎn)單realloc。我們的策略是先檢查頭部可回收區(qū)域尾部可寫空間的總和是否夠用。如果夠就執(zhí)行一次memmove將可讀數(shù)據(jù)移動(dòng)到緩沖區(qū)頭部騰出尾部空間。這稱為內(nèi)部騰挪。只有當(dāng)內(nèi)部騰挪后空間仍不足時(shí)才真正分配一塊更大的新內(nèi)存比如翻倍拷貝數(shù)據(jù)過去釋放舊內(nèi)存。避免抖動(dòng)擴(kuò)容因子如2倍要合適避免微小數(shù)據(jù)增長(zhǎng)導(dǎo)致頻繁擴(kuò)容。3. 核心接口設(shè)計(jì)與C實(shí)現(xiàn)細(xì)節(jié)有了設(shè)計(jì)思路我們來看具體實(shí)現(xiàn)。我將一個(gè)緩沖區(qū)類命名為Buffer它應(yīng)該提供以下幾組核心接口。3.1 構(gòu)造函數(shù)與內(nèi)存初始化class Buffer { public: static const size_t kCheapPrepend 8; // 預(yù)留空間可用于存消息長(zhǎng)度等 static const size_t kInitialSize 1024; // 初始可寫緩沖區(qū)大小 explicit Buffer(size_t initialSize kInitialSize) : buffer_(kCheapPrepend initialSize), // 總?cè)萘? 預(yù)留 初始 readerIndex_(kCheapPrepend), // 讀索引從預(yù)留后開始 writerIndex_(kCheapPrepend) { // 寫索引同上 assert(readableBytes() 0); assert(writableBytes() initialSize); assert(prependableBytes() kCheapPrepend); } // ... 其他成員函數(shù) private: std::vectorchar buffer_; // 底層存儲(chǔ) size_t readerIndex_; size_t writerIndex_; // 靜態(tài)斷言確保索引類型足夠大 static_assert(sizeof(readerIndex_) sizeof(size_t), 索引類型太小); };要點(diǎn)解析kCheapPrepend這是一個(gè)非常巧妙的設(shè)計(jì)。它在緩沖區(qū)最前面預(yù)留了一小段空間比如8字節(jié)。為什么在有些網(wǎng)絡(luò)協(xié)議中如自定義協(xié)議我們希望在應(yīng)用層報(bào)文前面加一個(gè)長(zhǎng)度字段。有了這個(gè)預(yù)留空間我們?cè)诮M裝消息時(shí)可以先把報(bào)文體append到緩沖區(qū)最后再計(jì)算長(zhǎng)度并prepend到預(yù)留空間里整個(gè)過程不需要移動(dòng)數(shù)據(jù)效率極高。使用std::vectorchar作為底層容器利用其RAII特性自動(dòng)管理內(nèi)存生命周期。初始狀態(tài)可讀數(shù)據(jù)為0可寫空間為initialSize可預(yù)留空間為kCheapPrepend。3.2 基礎(chǔ)狀態(tài)查詢接口這些接口是緩沖區(qū)運(yùn)轉(zhuǎn)的基礎(chǔ)必須高效inline且無誤。size_t readableBytes() const { return writerIndex_ - readerIndex_; } size_t writableBytes() const { return buffer_.size() - writerIndex_; } size_t prependableBytes() const { return readerIndex_; } // 頭部可回收/預(yù)留空間 const char* peek() const { return begin() readerIndex_; } // 獲取可讀數(shù)據(jù)首指針 char* beginWrite() { return begin() writerIndex_; } const char* beginWrite() const { return begin() writerIndex_; }begin()輔助函數(shù)返回buffer_的底層指針return *buffer_.begin();注意vector在為空時(shí)begin()可能未定義但我們構(gòu)造函數(shù)已分配內(nèi)存所以安全。3.3 核心操作retrieve,append,prepend這是緩沖區(qū)數(shù)據(jù)流動(dòng)的三大操作。1.retrieve消費(fèi)數(shù)據(jù)當(dāng)應(yīng)用層從緩沖區(qū)取走解析完len字節(jié)數(shù)據(jù)后需要移動(dòng)readerIndex_。void retrieve(size_t len) { if (len readableBytes()) { readerIndex_ len; // 消費(fèi)部分?jǐn)?shù)據(jù) } else { // len readableBytes() retrieveAll(); } } void retrieveAll() { readerIndex_ kCheapPrepend; writerIndex_ kCheapPrepend; }關(guān)鍵點(diǎn)retrieve并不真的刪除或清零數(shù)據(jù)只是移動(dòng)索引。這符合“零拷貝”思想。retrieveAll()非常高效直接將索引復(fù)位到初始狀態(tài)準(zhǔn)備接收新數(shù)據(jù)。2.append寫入數(shù)據(jù)這是最常用的操作將數(shù)據(jù)從外部寫入緩沖區(qū)的可寫區(qū)域。void append(const char* data, size_t len) { ensureWritableBytes(len); // 確?？臻g足夠可能觸發(fā)擴(kuò)容或騰挪 std::copy(data, data len, beginWrite()); hasWritten(len); // 移動(dòng)writerIndex_ } void ensureWritableBytes(size_t len) { if (writableBytes() len) { makeSpace(len); // 核心騰挪或擴(kuò)容 } } void hasWritten(size_t len) { writerIndex_ len; }makeSpace的實(shí)現(xiàn)是精華所在void makeSpace(size_t len) { // 情況1頭部預(yù)留空間尾部可寫空間足夠 if (writableBytes() prependableBytes() len kCheapPrepend) { // 不夠需要重新分配 buffer_.resize(writerIndex_ len); } else { // 夠進(jìn)行內(nèi)部騰挪 size_t readable readableBytes(); std::copy(begin() readerIndex_, begin() writerIndex_, begin() kCheapPrepend); readerIndex_ kCheapPrepend; writerIndex_ readerIndex_ readable; assert(readable readableBytes()); } }3.prepend向前寫入數(shù)據(jù)利用預(yù)留空間在可讀數(shù)據(jù)前面插入數(shù)據(jù)常用于添加協(xié)議頭。void prepend(const void* data, size_t len) { assert(len prependableBytes()); // 必須保證有足夠預(yù)留空間 readerIndex_ - len; const char* d static_castconst char*(data); std::copy(d, d len, begin() readerIndex_); }3.4 與網(wǎng)絡(luò)I/O的對(duì)接readFd與writeFd這是緩沖區(qū)價(jià)值體現(xiàn)的關(guān)鍵它封裝了與Socket文件描述符fd的交互。從Socket讀取數(shù)據(jù)readFd一個(gè)健壯的readFd需要處理TCP粘包、內(nèi)核緩沖區(qū)大小、以及EAGAIN/EWOULDBLOCK等問題。這里展示一個(gè)使用棧上額外緩沖區(qū)和readv的經(jīng)典高效實(shí)現(xiàn)// 返回讀取的字節(jié)數(shù)-1表示錯(cuò)誤errno需檢查0表示對(duì)端關(guān)閉 ssize_t Buffer::readFd(int fd, int* savedErrno) { // 棧上準(zhǔn)備一個(gè)額外緩沖區(qū)用于應(yīng)對(duì)緩沖區(qū)不夠的情況 char extrabuf[65536]; // 64K struct iovec vec[2]; const size_t writable writableBytes(); // 第一塊內(nèi)存Buffer自身的可寫空間 vec[0].iov_base beginWrite(); vec[0].iov_len writable; // 第二塊內(nèi)存棧上緩沖區(qū) vec[1].iov_base extrabuf; vec[1].iov_len sizeof(extrabuf); // 當(dāng)Buffer可寫空間較小時(shí)使用readv分散讀 const int iovcnt (writable sizeof(extrabuf)) ? 2 : 1; const ssize_t n ::readv(fd, vec, iovcnt); if (n 0) { *savedErrno errno; } else if (n writable) { // 數(shù)據(jù)全部讀到了Buffer的可寫空間 writerIndex_ n; } else { // 數(shù)據(jù)填滿了Buffer的可寫空間并有一部分讀到了extrabuf writerIndex_ buffer_.size(); // Buffer寫滿 append(extrabuf, n - writable); // 將extrabuf中的數(shù)據(jù)追加進(jìn)來會(huì)觸發(fā)擴(kuò)容 } return n; }為什么用readv和棧上緩沖區(qū)一次系統(tǒng)調(diào)用讀更多數(shù)據(jù)readv允許我們將數(shù)據(jù)分散讀到多個(gè)內(nèi)存塊。如果Buffer自身的空間足夠就只讀到Buffer里如果不夠剩余的數(shù)據(jù)可以讀到棧上的extrabuf避免因?yàn)锽uffer一時(shí)空間不足而丟失數(shù)據(jù)或觸發(fā)擴(kuò)容擴(kuò)容是堆操作更慢。棧上緩沖區(qū)速度快extrabuf在棧上分配和釋放極快。即使數(shù)據(jù)先到了extrabuf隨后再append到Buffer也多了一次內(nèi)存拷貝但這比因?yàn)锽uffer空間不足導(dǎo)致內(nèi)核數(shù)據(jù)無法及時(shí)讀出、觸發(fā)下一次EPOLLIN事件水平觸發(fā)模式下或等待邊緣觸發(fā)模式下要高效得多。這是一種用空間棧內(nèi)存換時(shí)間處理效率和編程復(fù)雜度的權(quán)衡。處理粘包readFd只負(fù)責(zé)把內(nèi)核緩沖區(qū)數(shù)據(jù)盡可能多地讀到應(yīng)用層緩沖區(qū)不負(fù)責(zé)解析消息邊界。消息邊界的解析如根據(jù)長(zhǎng)度字段或分隔符應(yīng)由應(yīng)用層在readFd之后通過peek()和retrieve()來完成。向Socket寫入數(shù)據(jù)writeFd將Buffer中可讀的數(shù)據(jù)通過Socket發(fā)送出去。// 返回寫入的字節(jié)數(shù)-1表示錯(cuò)誤需檢查errno ssize_t Buffer::writeFd(int fd, int* savedErrno) { size_t nLeft readableBytes(); ssize_t nWritten 0; const char* bufPtr peek(); while (nLeft 0) { nWritten ::write(fd, bufPtr, nLeft); if (nWritten 0) { if (errno EINTR) { // 被信號(hào)中斷重試 continue; } else if (errno EAGAIN || errno EWOULDBLOCK) { // 非阻塞模式下寫緩沖區(qū)已滿 break; } else { *savedErrno errno; // 其他錯(cuò)誤 return -1; } } nLeft - nWritten; bufPtr nWritten; retrieve(nWritten); // 重要發(fā)送成功的數(shù)據(jù)要從Buffer中消費(fèi)掉 } return (readableBytes() 0) ? 0 : (nWritten 0 ? nWritten : -1); // 返回0表示所有數(shù)據(jù)已寫完或EAGAIN-1表示錯(cuò)誤0表示部分寫入在非阻塞模式下可能發(fā)生 }關(guān)鍵點(diǎn)循環(huán)寫入因?yàn)閣rite系統(tǒng)調(diào)用可能只寫入部分?jǐn)?shù)據(jù)尤其是非阻塞Socket所以需要循環(huán)直到所有數(shù)據(jù)寫完或遇到EAGAIN。及時(shí)retrieve成功寫入多少字節(jié)就要從Buffer中消費(fèi)掉多少字節(jié)。否則下次writeFd會(huì)重復(fù)發(fā)送已發(fā)送的數(shù)據(jù)。錯(cuò)誤處理區(qū)分EINTR系統(tǒng)調(diào)用被信號(hào)中斷可重試、EAGAIN/EWOULDBLOCK非阻塞Socket寫緩沖區(qū)滿是正常情況應(yīng)停止循環(huán)等待下次可寫事件和其他致命錯(cuò)誤。4. 高級(jí)特性與性能優(yōu)化考量一個(gè)基礎(chǔ)的Buffer類已經(jīng)能工作但要用于生產(chǎn)環(huán)境還需要考慮更多。4.1 移動(dòng)語義與零拷貝優(yōu)化現(xiàn)代C強(qiáng)調(diào)移動(dòng)語義。我們的Buffer管理著vector應(yīng)該實(shí)現(xiàn)移動(dòng)構(gòu)造函數(shù)和移動(dòng)賦值運(yùn)算符避免不必要的深拷貝。Buffer(Buffer other) noexcept : buffer_(std::move(other.buffer_)), readerIndex_(other.readerIndex_), writerIndex_(other.writerIndex_) { other.readerIndex_ other.writerIndex_ kCheapPrepend; // 將other置于有效但空的狀態(tài) } Buffer operator(Buffer rhs) noexcept { if (this ! rhs) { buffer_ std::move(rhs.buffer_); readerIndex_ rhs.readerIndex_; writerIndex_ rhs.writerIndex_; rhs.readerIndex_ rhs.writerIndex_ kCheapPrepend; } return *this; } // 禁用拷貝構(gòu)造和拷貝賦值因?yàn)榭截惓杀靖咄ǔ２恍枰?Buffer(const Buffer) delete; Buffer operator(const Buffer) delete;對(duì)于某些極端性能場(chǎng)景可以考慮更激進(jìn)的零拷貝。例如如果有一個(gè)超大消息要發(fā)送可以不append到Buffer而是直接用一個(gè)std::vectorchar或std::string持有并記錄其生命周期在writeFd時(shí)使用writev將其與Buffer中的數(shù)據(jù)一并發(fā)送。但這會(huì)大大增加接口復(fù)雜性和生命周期管理的難度除非有確切的性能瓶頸否則不建議在基礎(chǔ)Buffer類中實(shí)現(xiàn)。4.2 線程安全性默認(rèn)情況下這個(gè)Buffer不是線程安全的。它被設(shè)計(jì)為每個(gè)TCP連接獨(dú)占一個(gè)Buffer實(shí)例而一個(gè)連接通常只在一個(gè)IO線程中被處理Reactor模式因此不需要內(nèi)部加鎖。如果你需要在多個(gè)線程間傳遞Buffer那么應(yīng)該由外層例如使用它的網(wǎng)絡(luò)庫或應(yīng)用通過消息隊(duì)列等同步機(jī)制來保證而不是在Buffer內(nèi)部加鎖。內(nèi)部加鎖會(huì)嚴(yán)重?fù)p害性能且鎖粒度難以控制。4.3 內(nèi)存池集成在高并發(fā)場(chǎng)景下即使有內(nèi)部騰挪頻繁的vector::resize涉及new/delete也可能成為瓶頸。一個(gè)優(yōu)化方向是集成內(nèi)存池Memory Pool。例如可以預(yù)分配一大塊內(nèi)存char數(shù)組然后以鏈表或數(shù)組管理多個(gè)固定大小的Buffer塊。當(dāng)Buffer需要擴(kuò)容時(shí)不是重新分配而是從內(nèi)存池中再申請(qǐng)一個(gè)塊并以鏈表形式連接起來這類似于std::deque但塊的大小是固定的、池化的。這能顯著減少系統(tǒng)調(diào)用malloc/free的次數(shù)但代價(jià)是Buffer不再是連續(xù)內(nèi)存peek()和beginWrite()可能失效需要更復(fù)雜的迭代器設(shè)計(jì)。這屬于高級(jí)優(yōu)化需要根據(jù)實(shí)際性能分析來決定是否引入。4.4 序列化與反序列化輔助Buffer可以作為簡(jiǎn)單的序列化工具。我們可以為其添加針對(duì)基本類型的append和retrieve重載。void appendInt32(int32_t x) { int32_t be32 htobe32(x); // 轉(zhuǎn)換為網(wǎng)絡(luò)字節(jié)序大端 append(be32, sizeof(be32)); } int32_t readInt32() { int32_t result peekInt32(); retrieve(sizeof(int32_t)); return result; } int32_t peekInt32() const { assert(readableBytes() sizeof(int32_t)); int32_t be32 0; ::memcpy(be32, peek(), sizeof(be32)); return be32toh(be32); // 轉(zhuǎn)換為主機(jī)字節(jié)序 }這樣應(yīng)用層協(xié)議編解碼會(huì)方便很多。注意字節(jié)序轉(zhuǎn)換htonl,ntohl等是網(wǎng)絡(luò)編程的必備步驟。5. 實(shí)戰(zhàn)應(yīng)用構(gòu)建一個(gè)簡(jiǎn)單的回顯服務(wù)器理論說再多不如看實(shí)際怎么用。我們用一個(gè)最簡(jiǎn)單的TCP回顯服務(wù)器Echo Server來演示Buffer的完整工作流程。這個(gè)服務(wù)器使用非阻塞IO和epollLinux或kqueueBSD的事件驅(qū)動(dòng)模型每個(gè)連接使用一個(gè)Buffer。核心事件循環(huán)偽代碼void onConnection(const TcpConnectionPtr conn) { if (conn-connected()) { // 為新連接分配一個(gè)Buffer conn-setContext(Buffer()); // 上下文綁定一個(gè)Buffer實(shí)例 } } void onMessage(const TcpConnectionPtr conn, Buffer* buf, Timestamp) { // buf 就是該連接對(duì)應(yīng)的Buffer // 1. 嘗試從buf中解析一條完整消息這里假設(shè)是換行符分隔 const char* crlf buf-findCRLF(); // 我們需要在Buffer里實(shí)現(xiàn)一個(gè)findCRLF方法 if (crlf) { // 2. 找到一條消息 size_t len crlf - buf-peek(); std::string message(buf-peek(), len); // 拷貝出來實(shí)際可零拷貝處理 buf-retrieve(len 2); // 消費(fèi)掉消息包括\r\n // 3. 處理消息這里簡(jiǎn)單回顯 LOG_INFO Echo message.size() bytes; conn-send(message \r\n); // conn-send內(nèi)部會(huì)調(diào)用我們Buffer的append } // 4. 如果buf中還有數(shù)據(jù)粘包下次onMessage會(huì)繼續(xù)處理 } void onWriteComplete(const TcpConnectionPtr conn) { // 數(shù)據(jù)發(fā)送完畢后的回調(diào)可用于流量控制等 }findCRLF的實(shí)現(xiàn)示例const char* Buffer::findCRLF() const { const char* crlf std::search(peek(), beginWrite(), kCRLF, kCRLF2); return crlf beginWrite() ? nullptr : crlf; } // 同理可以實(shí)現(xiàn)findEOL等在這個(gè)流程中onMessage被調(diào)用時(shí)連接對(duì)應(yīng)的Buffer里已經(jīng)由網(wǎng)絡(luò)庫通過readFd填入了最新的數(shù)據(jù)。我們嘗試從Buffer中查找消息邊界這里是\r\n。找到后取出消息內(nèi)容進(jìn)行處理回顯并調(diào)用retrieve消費(fèi)掉這部分?jǐn)?shù)據(jù)。如果沒找到完整消息說明數(shù)據(jù)還沒收全什么也不做等待下一次onMessage。發(fā)送數(shù)據(jù)時(shí)conn-send會(huì)將數(shù)據(jù)append到該連接的輸出緩沖區(qū)另一個(gè)Buffer實(shí)例然后網(wǎng)絡(luò)庫會(huì)在Socket可寫時(shí)調(diào)用該輸出緩沖區(qū)的writeFd將數(shù)據(jù)發(fā)送出去。這就是Buffer在網(wǎng)絡(luò)編程中的典型作用作為輸入輸出的數(shù)據(jù)暫存地和消息邊界解析器。6. 常見陷阱、調(diào)試技巧與性能測(cè)試即使設(shè)計(jì)再完善實(shí)際使用中還是會(huì)遇到各種問題。這里分享幾個(gè)我踩過的坑和解決方法。6.1 典型陷阱索引越界這是最常出現(xiàn)的bug。任何對(duì)readerIndex_和writerIndex_的修改都必須前置條件檢查assert在調(diào)試期很有用。確保readerIndex_ writerIndex_ buffer_.size()始終成立。忘記retrieve在writeFd成功發(fā)送數(shù)據(jù)后一定要調(diào)用retrieve。否則Buffer會(huì)認(rèn)為那些數(shù)據(jù)還沒被消費(fèi)導(dǎo)致內(nèi)存泄漏邏輯上和重復(fù)發(fā)送。擴(kuò)容策略激進(jìn)如果擴(kuò)容因子太大比如10倍可能一次性占用過多內(nèi)存。如果太小比如每次加100字節(jié)又會(huì)頻繁擴(kuò)容。需要根據(jù)業(yè)務(wù)流量特點(diǎn)進(jìn)行壓測(cè)和調(diào)整。線程安全誤用如前所述這個(gè)Buffer不是線程安全的。如果你在A線程append數(shù)據(jù)在B線程retrieve必須在外層加鎖。readFd中extrabuf大小?？臻g有限通常幾MBextrabuf不能太大如1MB否則有棧溢出風(fēng)險(xiǎn)。64KB是一個(gè)經(jīng)驗(yàn)值平衡了性能和安全性。6.2 調(diào)試技巧添加狀態(tài)打印函數(shù)在調(diào)試時(shí)一個(gè)dump()函數(shù)非常有用。void Buffer::dump() const { printf(Buffer size: %zu, readable: %zu, writable: %zu, prependable: %zu\n, buffer_.size(), readableBytes(), writableBytes(), prependableBytes()); printf(ReaderIndex: %zu, WriterIndex: %zu\n, readerIndex_, writerIndex_); // 可選以16進(jìn)制打印可讀數(shù)據(jù) for (size_t i readerIndex_; i writerIndex_; i) { printf(%02x , static_castunsigned char(buffer_[i])); if ((i - readerIndex_ 1) % 16 0) printf(\n); } printf(\n); }使用Valgrind或AddressSanitizer檢查內(nèi)存錯(cuò)誤如越界訪問、使用未初始化內(nèi)存等。壓力測(cè)試與內(nèi)存監(jiān)控編寫測(cè)試程序模擬海量連接和不同大小的數(shù)據(jù)包收發(fā)使用top、pmap或valgrind --toolmassif監(jiān)控內(nèi)存增長(zhǎng)和碎片情況。6.3 性能測(cè)試關(guān)注點(diǎn)設(shè)計(jì)一個(gè)簡(jiǎn)單的性能測(cè)試對(duì)比不同實(shí)現(xiàn)如直接read/write、使用std::string、使用我們的Buffer的差異。吞吐量Throughput在固定時(shí)間內(nèi)能處理多少數(shù)據(jù)。延遲Latency處理一個(gè)請(qǐng)求的平均時(shí)間。內(nèi)存占用Memory Footprint在處理大量連接時(shí)每個(gè)連接Buffer的內(nèi)存開銷。CPU使用率特別是在內(nèi)部騰挪和擴(kuò)容時(shí)的CPU消耗。測(cè)試時(shí)可以模擬不同的消息大小從幾個(gè)字節(jié)到幾十KB和發(fā)送頻率突發(fā)、勻速觀察Buffer的表現(xiàn)。通常我們的Buffer實(shí)現(xiàn)在處理大量小包和突發(fā)大包時(shí)會(huì)比樸素的方案穩(wěn)定得多。最后網(wǎng)絡(luò)緩沖區(qū)的設(shè)計(jì)沒有銀彈上述實(shí)現(xiàn)是一個(gè)在通用性、性能和復(fù)雜度之間取得較好平衡的方案。在實(shí)際項(xiàng)目中你可能還需要根據(jù)具體協(xié)議如HTTP/2的幀結(jié)構(gòu)、硬件特性如DPDK用戶態(tài)網(wǎng)絡(luò)進(jìn)行特化和優(yōu)化。但萬變不離其宗理解其核心設(shè)計(jì)思想——連續(xù)內(nèi)存、雙游標(biāo)、內(nèi)部騰挪、與I/O高效對(duì)接——將幫助你在面對(duì)任何網(wǎng)絡(luò)編程任務(wù)時(shí)都能設(shè)計(jì)出合適的數(shù)據(jù)緩沖區(qū)。