第2章-內存管理與Buffer封裝 · Lars-基于C++負載均衡遠程服務器調度系統

## 3) 內存管理與buffer封裝 ? 在完成網絡框架之前，我們先把必須的內存管理和buffer的封裝完成。這里我們先創建一個`io_buf`類，主要用來封裝基本的buffer結構。然后用一個`buf_pool`來管理全部的buffer集合。 ### 3.1 io_buf 內存塊 > lars_reactor/include/io_buf.h ```c #pragma once /* 定義一個 buffer存放數據的結構 * */ class io_buf { public: //構造，創建一個io_buf對象 io_buf(int size); //清空數據 void clear(); //將已經處理過的數據，清空,將未處理的數據提前至數據首地址 void adjust(); //將其他io_buf對象數據考本到自己中 void copy(const io_buf *other); //處理長度為len的數據，移動head和修正length void pop(int len); //如果存在多個buffer，是采用鏈表的形式鏈接起來 io_buf *next; //當前buffer的緩存容量大小 int capacity; //當前buffer有效數據長度 int length; //未處理數據的頭部位置索引 int head; //當前io_buf所保存的數據地址 char *data; }; ``` 對應的`io_buf`實現的文件,如下 > lars_reactor/src/io_buf.cpp ```c #include <stdio.h> #include <assert.h> #include <string.h> #include "io_buf.h" //構造，創建一個io_buf對象 io_buf::io_buf(int size): capacity(size), length(0), head(0), next(NULL) { data = new char[size]; assert(data); } //清空數據 void io_buf::clear() { length = head = 0; } //將已經處理過的數據，清空,將未處理的數據提前至數據首地址 void io_buf::adjust() { if (head != 0) { if (length != 0) { memmove(data, data+head, length); } head = 0; } } //將其他io_buf對象數據考本到自己中 void io_buf::copy(const io_buf *other) { memcpy(data, other->data + other->head, other->length); head = 0; length = other->length; } //處理長度為len的數據，移動head和修正length void io_buf::pop(int len) { length -= len; head += len; } ``` ? 這里主要要注意io_buf的兩個索引值length和head，一個是當前buffer的有效內存長度，haed則為可用的有效長度首數據位置。 capacity是io_buf的總容量空間大小。 ? 所以每次`pop()`則是彈出已經處理了多少，那么buffer剩下的內存就接下來需要處理的。 ? 然而`adjust()`則是從新重置io_buf,將所有數據都重新變成未處理狀態。 ? `clear()`則是將length和head清0，這里沒有提供`delete`真是刪除物理內存的方法，因為這里的buffer設計是不需要清理的，接下來是用一個`buf_pool`來管理全部未被使用的`io_buf`集合。而且`buf_pool`的管理的內存是程序開始預開辟的，不會做清理工作. ### 3.2 buf_pool 內存池 ? 接下來我們看看內存池的設計. > lars_reactor/include/buf_pool.h ```c #pragma once #include <ext/hash_map> #include "io_buf.h" typedef __gnu_cxx::hash_map<int, io_buf*> pool_t; enum MEM_CAP { m4K = 4096, m16K = 16384, m64K = 65536, m256K = 262144, m1M = 1048576, m4M = 4194304, m8M = 8388608 }; //總內存池最大限制單位是Kb 所以目前限制是 5GB #define EXTRA_MEM_LIMIT (5U *1024 *1024) /* * 定義buf內存池 * 設計為單例 * */ class buf_pool { public: //初始化單例對象 static void init() { //創建單例 _instance = new buf_pool(); } //獲取單例方法 static buf_pool *instance() { //保證init方法在這個進程執行中只被執行一次 pthread_once(&_once, init); return _instance; } //開辟一個io_buf io_buf *alloc_buf(int N); io_buf *alloc_buf() { return alloc_buf(m4K); } //重置一個io_buf void revert(io_buf *buffer); private: buf_pool(); //拷貝構造私有化 buf_pool(const buf_pool&); const buf_pool& operator=(const buf_pool&); //所有buffer的一個map集合句柄 pool_t _pool; //總buffer池的內存大小單位為KB uint64_t _total_mem; //單例對象 static buf_pool *_instance; //用于保證創建單例的init方法只執行一次的鎖 static pthread_once_t _once; //用戶保護內存池鏈表修改的互斥鎖 static pthread_mutex_t _mutex; }; ``` ? 首先`buf_pool`采用單例的方式進行設計。因為系統希望僅有一個內存池管理模塊。這里內存池用一個`__gnu_cxx::hash_map<int, io_buf*>`的map類型進行管理，其中key是每個組內存的空間容量，參考 ```c enum MEM_CAP { m4K = 4096, m16K = 16384, m64K = 65536, m256K = 262144, m1M = 1048576, m4M = 4194304, m8M = 8388608 }; ``` ? 其中每個key下面掛在一個`io_buf`鏈表。而且`buf_pool`預先會給map下的每個key的內存組開辟好一定數量的內存塊。然后上層用戶在使用的時候每次取出一個內存塊，就會將該內存塊從該內存組摘掉。當然使用完就放回來。如果不夠使用會額外開辟，也有最大的內存限制，在宏`EXTRA_MEM_LIMIT`中。具體的`buf_pool`實現如下: > lars_reactor/src/buf_pool.cpp ```c #include "buf_pool.h" #include <assert.h> //單例對象 buf_pool * buf_pool::_instance = NULL; //用于保證創建單例的init方法只執行一次的鎖 pthread_once_t buf_pool::_once = PTHREAD_ONCE_INIT; //用戶保護內存池鏈表修改的互斥鎖 pthread_mutex_t buf_pool::_mutex = PTHREAD_MUTEX_INITIALIZER; //構造函數主要是預先開辟一定量的空間 //這里buf_pool是一個hash，每個key都是不同空間容量 //對應的value是一個io_buf集合的鏈表 //buf_pool --> [m4K] -- io_buf-io_buf-io_buf-io_buf... // [m16K] -- io_buf-io_buf-io_buf-io_buf... // [m64K] -- io_buf-io_buf-io_buf-io_buf... // [m256K] -- io_buf-io_buf-io_buf-io_buf... // [m1M] -- io_buf-io_buf-io_buf-io_buf... // [m4M] -- io_buf-io_buf-io_buf-io_buf... // [m8M] -- io_buf-io_buf-io_buf-io_buf... buf_pool::buf_pool():_total_mem(0) { io_buf *prev; //----> 開辟4K buf 內存池 _pool[m4K] = new io_buf(m4K); if (_pool[m4K] == NULL) { fprintf(stderr, "new io_buf m4K error"); exit(1); } prev = _pool[m4K]; //4K的io_buf 預先開辟5000個，約20MB供開發者使用 for (int i = 1; i < 5000; i ++) { prev->next = new io_buf(m4K); if (prev->next == NULL) { fprintf(stderr, "new io_buf m4K error"); exit(1); } prev = prev->next; } _total_mem += 4 * 5000; //----> 開辟16K buf 內存池 _pool[m16K] = new io_buf(m16K); if (_pool[m16K] == NULL) { fprintf(stderr, "new io_buf m16K error"); exit(1); } prev = _pool[m16K]; //16K的io_buf 預先開辟1000個，約16MB供開發者使用 for (int i = 1; i < 1000; i ++) { prev->next = new io_buf(m16K); if (prev->next == NULL) { fprintf(stderr, "new io_buf m16K error"); exit(1); } prev = prev->next; } _total_mem += 16 * 1000; //----> 開辟64K buf 內存池 _pool[m64K] = new io_buf(m64K); if (_pool[m64K] == NULL) { fprintf(stderr, "new io_buf m64K error"); exit(1); } prev = _pool[m64K]; //64K的io_buf 預先開辟500個，約32MB供開發者使用 for (int i = 1; i < 500; i ++) { prev->next = new io_buf(m64K); if (prev->next == NULL) { fprintf(stderr, "new io_buf m64K error"); exit(1); } prev = prev->next; } _total_mem += 64 * 500; //----> 開辟256K buf 內存池 _pool[m256K] = new io_buf(m256K); if (_pool[m256K] == NULL) { fprintf(stderr, "new io_buf m256K error"); exit(1); } prev = _pool[m256K]; //256K的io_buf 預先開辟200個，約50MB供開發者使用 for (int i = 1; i < 200; i ++) { prev->next = new io_buf(m256K); if (prev->next == NULL) { fprintf(stderr, "new io_buf m256K error"); exit(1); } prev = prev->next; } _total_mem += 256 * 200; //----> 開辟1M buf 內存池 _pool[m1M] = new io_buf(m1M); if (_pool[m1M] == NULL) { fprintf(stderr, "new io_buf m1M error"); exit(1); } prev = _pool[m1M]; //1M的io_buf 預先開辟50個，約50MB供開發者使用 for (int i = 1; i < 50; i ++) { prev->next = new io_buf(m1M); if (prev->next == NULL) { fprintf(stderr, "new io_buf m1M error"); exit(1); } prev = prev->next; } _total_mem += 1024 * 50; //----> 開辟4M buf 內存池 _pool[m4M] = new io_buf(m4M); if (_pool[m4M] == NULL) { fprintf(stderr, "new io_buf m4M error"); exit(1); } prev = _pool[m4M]; //4M的io_buf 預先開辟20個，約80MB供開發者使用 for (int i = 1; i < 20; i ++) { prev->next = new io_buf(m4M); if (prev->next == NULL) { fprintf(stderr, "new io_buf m4M error"); exit(1); } prev = prev->next; } _total_mem += 4096 * 20; //----> 開辟8M buf 內存池 _pool[m8M] = new io_buf(m8M); if (_pool[m8M] == NULL) { fprintf(stderr, "new io_buf m8M error"); exit(1); } prev = _pool[m8M]; //8M的io_buf 預先開辟10個，約80MB供開發者使用 for (int i = 1; i < 10; i ++) { prev->next = new io_buf(m8M); if (prev->next == NULL) { fprintf(stderr, "new io_buf m8M error"); exit(1); } prev = prev->next; } _total_mem += 8192 * 10; } //開辟一個io_buf //1 如果上層需要N個字節的大小的空間，找到與N最接近的buf hash組，取出， //2 如果該組已經沒有節點使用，可以額外申請 //3 總申請長度不能夠超過最大的限制大小 EXTRA_MEM_LIMIT //4 如果有該節點需要的內存塊，直接取出，并且將該內存塊從pool摘除 io_buf *buf_pool::alloc_buf(int N) { //1 找到N最接近哪hash 組 int index; if (N <= m4K) { index = m4K; } else if (N <= m16K) { index = m16K; } else if (N <= m64K) { index = m64K; } else if (N <= m256K) { index = m256K; } else if (N <= m1M) { index = m1M; } else if (N <= m4M) { index = m4M; } else if (N <= m8M) { index = m8M; } else { return NULL; } //2 如果該組已經沒有，需要額外申請，那么需要加鎖保護 pthread_mutex_lock(&_mutex); if (_pool[index] == NULL) { if (_total_mem + index/1024 >= EXTRA_MEM_LIMIT) { //當前的開辟的空間已經超過最大限制 fprintf(stderr, "already use too many memory!\n"); exit(1); } io_buf *new_buf = new io_buf(index); if (new_buf == NULL) { fprintf(stderr, "new io_buf error\n"); exit(1); } _total_mem += index/1024; pthread_mutex_unlock(&_mutex); return new_buf; } //3 從pool中摘除該內存塊 io_buf *target = _pool[index]; _pool[index] = target->next; pthread_mutex_unlock(&_mutex); target->next = NULL; return target; } //重置一個io_buf,將一個buf 上層不再使用，或者使用完成之后，需要將該buf放回pool中 void buf_pool::revert(io_buf *buffer) { //每個buf的容量都是固定的在hash的key中取值 int index = buffer->capacity; //重置io_buf中的內置位置指針 buffer->length = 0; buffer->head = 0; pthread_mutex_lock(&_mutex); //找到對應的hash組 buf首屆點地址 assert(_pool.find(index) != _pool.end()); //將buffer插回鏈表頭部 buffer->next = _pool[index]; _pool[index] = buffer; pthread_mutex_unlock(&_mutex); } ``` ? 其中，`buf_pool`構造函數中實現了內存池的hash預開辟內存工作，具體的數據結構如下 ```c //buf_pool --> [m4K] --> io_buf-io_buf-io_buf-io_buf... // [m16K] --> io_buf-io_buf-io_buf-io_buf... // [m64K] --> io_buf-io_buf-io_buf-io_buf... // [m256K] --> io_buf-io_buf-io_buf-io_buf... // [m1M] --> io_buf-io_buf-io_buf-io_buf... // [m4M] --> io_buf-io_buf-io_buf-io_buf... // [m8M] --> io_buf-io_buf-io_buf-io_buf... ``` ? `alloc_buf()`方法，是調用者從內存池中取出一塊內存，如果最匹配的內存塊存在，則返回，并將該塊內存從buf_pool中摘除掉，如果沒有則開辟一個內存出來。 `revert()`方法則是將已經使用完的`io_buf`重新放回`buf_pool`中。 ### 3.3 讀寫buffer機制 ? 那么接下來我們就需要實現一個專門用來讀(輸入)數據的`input_buf`和專門用來寫(輸出)數據的`output_buf`類了。由于這兩個人都應該擁有一些`io_buf`的特性，所以我們先定義一個基礎的父類`reactor_buf`。 #### A. reactor_buf類 > lars_reactor/include/reactor_buf.h ```c #pragma once #include "io_buf.h" #include "buf_pool.h" #include <assert.h> #include <unistd.h> /* * 給業務層提供的最后tcp_buffer結構 * */ class reactor_buf { public: reactor_buf(); ~reactor_buf(); const int length() const; void pop(int len); void clear(); protected: io_buf *_buf; }; ``` ? 這個的作用就是將io_buf作為自己的一個成員，然后做了一些包裝。具體方法實現如下。 > lars_reactor/src/reactor.cpp ```c #include "reactor_buf.h" #include <sys/ioctl.h> #include <string.h> reactor_buf::reactor_buf() { _buf = NULL; } reactor_buf::~reactor_buf() { clear(); } const int reactor_buf::length() const { return _buf != NULL? _buf->length : 0; } void reactor_buf::pop(int len) { assert(_buf != NULL && len <= _buf->length); _buf->pop(len); //當此時_buf的可用長度已經為0 if(_buf->length == 0) { //將_buf重新放回buf_pool中 buf_pool::instance()->revert(_buf); _buf = NULL; } } void reactor_buf::clear() { if (_buf != NULL) { //將_buf重新放回buf_pool中 buf_pool::instance()->revert(_buf); _buf = NULL; } } ``` #### B. input_buf類 ? 接下來就可以集成`reactor_buf`類實現`input_buf`類的設計了。 > lars_reactor/include/reactor_buf.h ```h //讀(輸入) 緩存buffer class input_buf : public reactor_buf { public: //從一個fd中讀取數據到reactor_buf中 int read_data(int fd); //取出讀到的數據 const char *data() const; //重置緩沖區 void adjust(); }; ``` ? 其中data()方法即取出已經讀取的數據，adjust()含義和`io_buf`含義一致。主要是`read_data()`方法。具體實現如下。 > lars_reactor/src/reactor.cpp ```c //從一個fd中讀取數據到reactor_buf中 int input_buf::read_data(int fd) { int need_read;//硬件有多少數據可以讀 //一次性讀出所有的數據 //需要給fd設置FIONREAD, //得到read緩沖中有多少數據是可以讀取的 if (ioctl(fd, FIONREAD, &need_read) == -1) { fprintf(stderr, "ioctl FIONREAD\n"); return -1; } if (_buf == NULL) { //如果io_buf為空,從內存池申請 _buf = buf_pool::instance()->alloc_buf(need_read); if (_buf == NULL) { fprintf(stderr, "no idle buf for alloc\n"); return -1; } } else { //如果io_buf可用，判斷是否夠存 assert(_buf->head == 0); if (_buf->capacity - _buf->length < (int)need_read) { //不夠存，沖內存池申請 io_buf *new_buf = buf_pool::instance()->alloc_buf(need_read+_buf->length); if (new_buf == NULL) { fprintf(stderr, "no ilde buf for alloc\n"); return -1; } //將之前的_buf的數據考到新申請的buf中 new_buf->copy(_buf); //將之前的_buf放回內存池中 buf_pool::instance()->revert(_buf); //新申請的buf成為當前io_buf _buf = new_buf; } } //讀取數據 int already_read = 0; do { //讀取的數據拼接到之前的數據之后 if(need_read == 0) { //可能是read阻塞讀數據的模式，對方未寫數據 already_read = read(fd, _buf->data + _buf->length, m4K); } else { already_read = read(fd, _buf->data + _buf->length, need_read); } } while (already_read == -1 && errno == EINTR); //systemCall引起的中斷繼續讀取 if (already_read > 0) { if (need_read != 0) { assert(already_read == need_read); } _buf->length += already_read; } return already_read; } //取出讀到的數據 const char *input_buf::data() const { return _buf != NULL ? _buf->data + _buf->head : NULL; } //重置緩沖區 void input_buf::adjust() { if (_buf != NULL) { _buf->adjust(); } } ``` #### C. output_buf類 ? 接下來就可以集成`reactor_buf`類實現`output_buf`類的設計了。 > lars_reactor/include/reactor_buf.h ```h //寫(輸出) 緩存buffer class output_buf : public reactor_buf { public: //將一段數據寫到一個reactor_buf中 int send_data(const char *data, int datalen); //將reactor_buf中的數據寫到一個fd中 int write2fd(int fd); }; ``` ? `send_data()`方法主要是將數據寫到`io_buf`中，實際上并沒有做真正的寫操作。而是當調用`write2fd`方法時，才會將`io_buf`的數據寫到對應的fd中。send_data是做一些buf內存塊的申請等工作。具體實現如下 > lars_reactor/src/reactor.cpp ```c //將一段數據寫到一個reactor_buf中 int output_buf::send_data(const char *data, int datalen) { if (_buf == NULL) { //如果io_buf為空,從內存池申請 _buf = buf_pool::instance()->alloc_buf(datalen); if (_buf == NULL) { fprintf(stderr, "no idle buf for alloc\n"); return -1; } } else { //如果io_buf可用，判斷是否夠存 assert(_buf->head == 0); if (_buf->capacity - _buf->length < datalen) { //不夠存，沖內存池申請 io_buf *new_buf = buf_pool::instance()->alloc_buf(datalen+_buf->length); if (new_buf == NULL) { fprintf(stderr, "no ilde buf for alloc\n"); return -1; } //將之前的_buf的數據考到新申請的buf中 new_buf->copy(_buf); //將之前的_buf放回內存池中 buf_pool::instance()->revert(_buf); //新申請的buf成為當前io_buf _buf = new_buf; } } //將data數據拷貝到io_buf中,拼接到后面 memcpy(_buf->data + _buf->length, data, datalen); _buf->length += datalen; return 0; } //將reactor_buf中的數據寫到一個fd中 int output_buf::write2fd(int fd) { assert(_buf != NULL && _buf->head == 0); int already_write = 0; do { already_write = write(fd, _buf->data, _buf->length); } while (already_write == -1 && errno == EINTR); //systemCall引起的中斷，繼續寫 if (already_write > 0) { //已經處理的數據清空 _buf->pop(already_write); //未處理數據前置，覆蓋老數據 _buf->adjust(); } //如果fd非阻塞，可能會得到EAGAIN錯誤 if (already_write == -1 && errno == EAGAIN) { already_write = 0;//不是錯誤，僅僅返回0，表示目前是不可以繼續寫的 } return already_write; } ``` ? 現在我們已經完成了內存管理及讀寫buf機制的實現，接下來就要簡單的測試一下，用我們之前的V0.1版本的reactor server來測試。 ### 3.4 完成Lars Reactor V0.2開發 #### A. 修改tcp_server ? 主要修正do_accept()方法，加上reactor_buf機制. > lars_reactor/src/tcp_server.cpp ```c #include <stdio.h> #include <stdlib.h> #include <string.h> #include <strings.h> #include <unistd.h> #include <signal.h> #include <sys/types.h> /* See NOTES */ #include <sys/socket.h> #include <arpa/inet.h> #include <errno.h> #include "tcp_server.h" #include "reactor_buf.h" //server的構造函數 tcp_server::tcp_server(const char *ip, uint16_t port) { //... } //開始提供創建鏈接服務 void tcp_server::do_accept() { int connfd; while(true) { //accept與客戶端創建鏈接 printf("begin accept\n"); connfd = accept(_sockfd, (struct sockaddr*)&_connaddr, &_addrlen); if (connfd == -1) { if (errno == EINTR) { fprintf(stderr, "accept errno=EINTR\n"); continue; } else if (errno == EMFILE) { //建立鏈接過多，資源不夠 fprintf(stderr, "accept errno=EMFILE\n"); } else if (errno == EAGAIN) { fprintf(stderr, "accept errno=EAGAIN\n"); break; } else { fprintf(stderr, "accept error"); exit(1); } } else { //accept succ! int ret = 0; input_buf ibuf; output_buf obuf; char *msg = NULL; int msg_len = 0; do { ret = ibuf.read_data(connfd); if (ret == -1) { fprintf(stderr, "ibuf read_data error\n"); break; } printf("ibuf.length() = %d\n", ibuf.length()); //將讀到的數據放在msg中 msg_len = ibuf.length(); msg = (char*)malloc(msg_len); bzero(msg, msg_len); memcpy(msg, ibuf.data(), msg_len); ibuf.pop(msg_len); ibuf.adjust(); printf("recv data = %s\n", msg); //回顯數據 obuf.send_data(msg, msg_len); while(obuf.length()) { int write_ret = obuf.write2fd(connfd); if (write_ret == -1) { fprintf(stderr, "write connfd error\n"); return; } else if(write_ret == 0) { //不是錯誤，表示此時不可寫 break; } } free(msg); } while (ret != 0); //Peer is closed close(connfd); } } } ``` 編譯生成新的liblreactor.a ```bash $cd lars_reactor/ $make g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/tcp_server.o src/tcp_server.cpp -I./include g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/io_buf.o src/io_buf.cpp -I./include g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/reactor_buf.o src/reactor_buf.cpp -I./include g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/buf_pool.o src/buf_pool.cpp -I./include mkdir -p lib ar cqs lib/liblreactor.a src/tcp_server.o src/io_buf.o src/reactor_buf.o src/buf_pool.o ``` #### B. 編譯V0.2 server APP ? 我們將lars_reactor/example/lars_reactor_0.1 的代碼復制一份到 lars_reactor/example/lars_reactor_0.2中。由于我們這里使用了pthread庫，所以在lars_reactor_0.2的Makefile文件要加上pthread庫的關聯 > lars_reactor/example/lars_reactor_0.2/Makefile ```makefile CXX=g++ CFLAGS=-g -O2 -Wall -fPIC -Wno-deprecated INC=-I../../include LIB=-L../../lib -llreactor -lpthread OBJS = $(addsuffix .o, $(basename $(wildcard *.cc))) all: $(CXX) -o lars_reactor $(CFLAGS) lars_reactor.cpp $(INC) $(LIB) clean: -rm -f *.o lars_reactor ``` 編譯在lars_reactor/example/lars_reactor_0.2/ ```bash $ cd lars_reactor/example/lars_reactor_0.2/ $ make g++ -o lars_reactor -g -O2 -Wall -fPIC -Wno-deprecated lars_reactor.cpp -I../../include -L../../lib -llreactor -lpthread ``` #### C. 測試啟動server ```bash $ ./lars_reactor begin accept ``` 啟動client ```bash $ nc 127.0.0.1 7777 ``` 客戶端輸入文字，效果如下：服務端： ```bash ibuf.length() = 21 recv data = hello lars, By Aceld ``` 客戶端： ```bash $ nc 127.0.0.1 7777 hello lars, By Aceld hello lars, By Aceld ``` ? ok!現在我們的讀寫buffer機制已經成功的集成到我們的lars網絡框架中了。 --- ### 關于作者：作者：`Aceld(劉丹冰)` mail: [danbing.at@gmail.com](mailto:danbing.at@gmail.com) github: [https://github.com/aceld](https://github.com/aceld) 原創書籍: [http://www.hmoore.net/@aceld](http://www.hmoore.net/@aceld) ![](https://img.kancloud.cn/b0/d1/b0d11a21ba62e96aef1c11d5bfff2cf8_227x227.jpg) >**原創聲明:未經作者允許請勿轉載, 如果轉載請注明出處**