3) 内存管理与buffer封装

  1. 在完成网络框架之前，我们先把必须的内存管理和buffer的封装完成。

这里我们先创建一个io_buf类，主要用来封装基本的buffer结构。然后用一个buf_pool来管理全部的buffer集合。

3.1 io_buf 内存块

lars_reactor/include/io_buf.h

  1. #pragma once
  3. /*
  4.     定义一个 buffer存放数据的结构
  5.  * */
  6. class io_buf {
  7. public:
  8.     //构造，创建一个io_buf对象
  9.     io_buf(int size);
  11.     //清空数据
  12.     void clear();
  14.     //将已经处理过的数据，清空,将未处理的数据提前至数据首地址
  15.     void adjust();
  17.     //将其他io_buf对象数据考本到自己中
  18.     void copy(const io_buf *other);
  20.     //处理长度为len的数据，移动head和修正length
  21.     void pop(int len);
  23.     //如果存在多个buffer，是采用链表的形式链接起来
  24.     io_buf *next;
  26.     //当前buffer的缓存容量大小
  27.     int capacity;
  28.     //当前buffer有效数据长度
  29.     int length;
  30.     //未处理数据的头部位置索引
  31.     int head;
  32.     //当前io_buf所保存的数据地址
  33.     char *data;
  34. };

对应的io_buf实现的文件,如下

lars_reactor/src/io_buf.cpp

  1. #include <stdio.h>
  2. #include <assert.h>
  3. #include <string.h>
  4. #include "io_buf.h"
  6. //构造，创建一个io_buf对象
  7. io_buf::io_buf(int size):
  8. capacity(size), 
  9. length(0),
  10. head(0),
  11. next(NULL) 
  12. {
  13.    data = new char[size];
  14.    assert(data);
  15. }
  17. //清空数据
  18. void io_buf::clear() {
  19.     length = head = 0;
  20. }
  22. //将已经处理过的数据，清空,将未处理的数据提前至数据首地址
  23. void io_buf::adjust() {
  24.     if (head != 0) {
  25.         if (length != 0) {
  26.             memmove(data, data+head, length);
  27.         }
  28.         head = 0;
  29.     }
  30. }
  32. //将其他io_buf对象数据考本到自己中
  33. void io_buf::copy(const io_buf *other) {
  34.     memcpy(data, other->data + other->head, other->length);
  35.     head = 0;
  36.     length = other->length;
  37. }
  39. //处理长度为len的数据，移动head和修正length
  40. void io_buf::pop(int len) {
  41.     length -= len;
  42.     head += len;
  43. }
  1. 这里主要要注意io_buf的两个索引值length和head，一个是当前buffer的有效内存长度，haed则为可用的有效长度首数据位置。 capacity是io_buf的总容量空间大小。
  3. 所以每次`pop()`则是弹出已经处理了多少，那么buffer剩下的内存就接下来需要处理的。
  5. 然而`adjust()`则是从新重置io_buf,将所有数据都重新变成未处理状态。
  7. `clear()`则是将length和head清0，这里没有提供`delete`真是删除物理内存的方法，因为这里的buffer设计是不需要清理的，接下来是用一个`buf_pool`来管理全部未被使用的`io_buf`集合。而且`buf_pool`的管理的内存是程序开始预开辟的，不会做清理工作.

3.2 buf_pool 内存池

  1. 接下来我们看看内存池的设计.

lars_reactor/include/buf_pool.h

  1. #pragma once
  3. #include <ext/hash_map>
  4. #include "io_buf.h"
  6. typedef __gnu_cxx::hash_map<int, io_buf*> pool_t;
  8. enum MEM_CAP {
  9.     m4K     = 4096,
  10.     m16K    = 16384,
  11.     m64K    = 65536,
  12.     m256K   = 262144,
  13.     m1M     = 1048576,
  14.     m4M     = 4194304,
  15.     m8M     = 8388608
  16. };
  19. //总内存池最大限制 单位是Kb 所以目前限制是 5GB
  20. #define EXTRA_MEM_LIMIT (5U *1024 *1024) 
  22. /*
  23.  *  定义buf内存池
  24.  *  设计为单例
  25.  * */
  26. class buf_pool 
  27. {
  28. public:
  29.     //初始化单例对象
  30.     static void init() {
  31.         //创建单例
  32.         _instance = new buf_pool();
  33.     }
  35.     //获取单例方法
  36.     static buf_pool *instance() {
  37.         //保证init方法在这个进程执行中 只被执行一次
  38.         pthread_once(&_once, init);
  39.         return _instance;
  40.     }
  42.     //开辟一个io_buf
  43.     io_buf *alloc_buf(int N);
  44.     io_buf *alloc_buf() { return alloc_buf(m4K); }
  47.     //重置一个io_buf
  48.     void revert(io_buf *buffer);
  51. private:
  52.     buf_pool();
  54.     //拷贝构造私有化
  55.     buf_pool(const buf_pool&);
  56.     const buf_pool& operator=(const buf_pool&);
  58.     //所有buffer的一个map集合句柄
  59.     pool_t _pool;
  61.     //总buffer池的内存大小 单位为KB
  62.     uint64_t _total_mem;
  64.     //单例对象
  65.     static buf_pool *_instance;
  67.     //用于保证创建单例的init方法只执行一次的锁
  68.     static pthread_once_t _once;
  70.     //用户保护内存池链表修改的互斥锁
  71.     static pthread_mutex_t _mutex;
  72. };
  1.     首先`buf_pool`采用单例的方式进行设计。因为系统希望仅有一个内存池管理模块。这里内存池用一个`__gnu_cxx::hash_map<int, io_buf*>`的map类型进行管理，其中key是每个组内存的空间容量，参考
  1. enum MEM_CAP {
  2.     m4K     = 4096,
  3.     m16K    = 16384,
  4.     m64K    = 65536,
  5.     m256K   = 262144,
  6.     m1M     = 1048576,
  7.     m4M     = 4194304,
  8.     m8M     = 8388608
  9. };
  1.     其中每个key下面挂在一个`io_buf`链表。而且`buf_pool`预先会给map下的每个key的内存组开辟好一定数量的内存块。然后上层用户在使用的时候每次取出一个内存块，就会将该内存块从该内存组摘掉。当然使用完就放回来。如果不够使用会额外开辟，也有最大的内存限制，在宏`EXTRA_MEM_LIMIT`中。

具体的buf_pool实现如下:

lars_reactor/src/buf_pool.cpp

  1. #include "buf_pool.h"
  2. #include <assert.h>
  5. //单例对象
  6. buf_pool * buf_pool::_instance = NULL;
  8. //用于保证创建单例的init方法只执行一次的锁
  9. pthread_once_t buf_pool::_once = PTHREAD_ONCE_INIT;
  12. //用户保护内存池链表修改的互斥锁
  13. pthread_mutex_t buf_pool::_mutex = PTHREAD_MUTEX_INITIALIZER;
  16. //构造函数 主要是预先开辟一定量的空间
  17. //这里buf_pool是一个hash，每个key都是不同空间容量
  18. //对应的value是一个io_buf集合的链表
  19. //buf_pool -->  [m4K] -- io_buf-io_buf-io_buf-io_buf...
  20. //              [m16K] -- io_buf-io_buf-io_buf-io_buf...
  21. //              [m64K] -- io_buf-io_buf-io_buf-io_buf...
  22. //              [m256K] -- io_buf-io_buf-io_buf-io_buf...
  23. //              [m1M] -- io_buf-io_buf-io_buf-io_buf...
  24. //              [m4M] -- io_buf-io_buf-io_buf-io_buf...
  25. //              [m8M] -- io_buf-io_buf-io_buf-io_buf...
  26. buf_pool::buf_pool():_total_mem(0)
  27. {
  28.     io_buf *prev; 
  30.     //----> 开辟4K buf 内存池
  31.     _pool[m4K] = new io_buf(m4K);
  32.     if (_pool[m4K] == NULL) {
  33.         fprintf(stderr, "new io_buf m4K error");
  34.         exit(1);
  35.     }
  37.     prev = _pool[m4K];
  38.     //4K的io_buf 预先开辟5000个，约20MB供开发者使用
  39.     for (int i = 1; i < 5000; i ++) {
  40.         prev->next = new io_buf(m4K);
  41.         if (prev->next == NULL) {
  42.             fprintf(stderr, "new io_buf m4K error");
  43.             exit(1);
  44.         }
  45.         prev = prev->next;
  46.     }
  47.     _total_mem += 4 * 5000;
  51.     //----> 开辟16K buf 内存池
  52.     _pool[m16K] = new io_buf(m16K);
  53.     if (_pool[m16K] == NULL) {
  54.         fprintf(stderr, "new io_buf m16K error");
  55.         exit(1);
  56.     }
  58.     prev = _pool[m16K];
  59.     //16K的io_buf 预先开辟1000个，约16MB供开发者使用
  60.     for (int i = 1; i < 1000; i ++) {
  61.         prev->next = new io_buf(m16K);
  62.         if (prev->next == NULL) {
  63.             fprintf(stderr, "new io_buf m16K error");
  64.             exit(1);
  65.         }
  66.         prev = prev->next;
  67.     }
  68.     _total_mem += 16 * 1000;
  72.     //----> 开辟64K buf 内存池
  73.     _pool[m64K] = new io_buf(m64K);
  74.     if (_pool[m64K] == NULL) {
  75.         fprintf(stderr, "new io_buf m64K error");
  76.         exit(1);
  77.     }
  79.     prev = _pool[m64K];
  80.     //64K的io_buf 预先开辟500个，约32MB供开发者使用
  81.     for (int i = 1; i < 500; i ++) {
  82.         prev->next = new io_buf(m64K);
  83.         if (prev->next == NULL) {
  84.             fprintf(stderr, "new io_buf m64K error");
  85.             exit(1);
  86.         }
  87.         prev = prev->next;
  88.     }
  89.     _total_mem += 64 * 500;
  92.     //----> 开辟256K buf 内存池
  93.     _pool[m256K] = new io_buf(m256K);
  94.     if (_pool[m256K] == NULL) {
  95.         fprintf(stderr, "new io_buf m256K error");
  96.         exit(1);
  97.     }
  99.     prev = _pool[m256K];
  100.     //256K的io_buf 预先开辟200个，约50MB供开发者使用
  101.     for (int i = 1; i < 200; i ++) {
  102.         prev->next = new io_buf(m256K);
  103.         if (prev->next == NULL) {
  104.             fprintf(stderr, "new io_buf m256K error");
  105.             exit(1);
  106.         }
  107.         prev = prev->next;
  108.     }
  109.     _total_mem += 256 * 200;
  112.     //----> 开辟1M buf 内存池
  113.     _pool[m1M] = new io_buf(m1M);
  114.     if (_pool[m1M] == NULL) {
  115.         fprintf(stderr, "new io_buf m1M error");
  116.         exit(1);
  117.     }
  119.     prev = _pool[m1M];
  120.     //1M的io_buf 预先开辟50个，约50MB供开发者使用
  121.     for (int i = 1; i < 50; i ++) {
  122.         prev->next = new io_buf(m1M);
  123.         if (prev->next == NULL) {
  124.             fprintf(stderr, "new io_buf m1M error");
  125.             exit(1);
  126.         }
  127.         prev = prev->next;
  128.     }
  129.     _total_mem += 1024 * 50;
  132.     //----> 开辟4M buf 内存池
  133.     _pool[m4M] = new io_buf(m4M);
  134.     if (_pool[m4M] == NULL) {
  135.         fprintf(stderr, "new io_buf m4M error");
  136.         exit(1);
  137.     }
  139.     prev = _pool[m4M];
  140.     //4M的io_buf 预先开辟20个，约80MB供开发者使用
  141.     for (int i = 1; i < 20; i ++) {
  142.         prev->next = new io_buf(m4M);
  143.         if (prev->next == NULL) {
  144.             fprintf(stderr, "new io_buf m4M error");
  145.             exit(1);
  146.         }
  147.         prev = prev->next;
  148.     }
  149.     _total_mem += 4096 * 20;
  153.     //----> 开辟8M buf 内存池
  154.     _pool[m8M] = new io_buf(m8M);
  155.     if (_pool[m8M] == NULL) {
  156.         fprintf(stderr, "new io_buf m8M error");
  157.         exit(1);
  158.     }
  160.     prev = _pool[m8M];
  161.     //8M的io_buf 预先开辟10个，约80MB供开发者使用
  162.     for (int i = 1; i < 10; i ++) {
  163.         prev->next = new io_buf(m8M);
  164.         if (prev->next == NULL) {
  165.             fprintf(stderr, "new io_buf m8M error");
  166.             exit(1);
  167.         }
  168.         prev = prev->next;
  169.     }
  170.     _total_mem += 8192 * 10;
  171. }
  174. //开辟一个io_buf
  175. //1 如果上层需要N个字节的大小的空间，找到与N最接近的buf hash组，取出，
  176. //2 如果该组已经没有节点使用，可以额外申请
  177. //3 总申请长度不能够超过最大的限制大小 EXTRA_MEM_LIMIT
  178. //4 如果有该节点需要的内存块，直接取出，并且将该内存块从pool摘除
  179. io_buf *buf_pool::alloc_buf(int N) 
  180. {
  181.     //1 找到N最接近哪hash 组
  182.     int index;
  183.     if (N <= m4K) {
  184.         index = m4K;
  185.     }
  186.     else if (N <= m16K) {
  187.         index = m16K;
  188.     }
  189.     else if (N <= m64K) {
  190.         index = m64K;
  191.     }
  192.     else if (N <= m256K) {
  193.         index = m256K;
  194.     }
  195.     else if (N <= m1M) {
  196.         index = m1M;
  197.     }
  198.     else if (N <= m4M) {
  199.         index = m4M;
  200.     }
  201.     else if (N <= m8M) {
  202.         index = m8M;
  203.     }
  204.     else {
  205.         return NULL;
  206.     }
  209.     //2 如果该组已经没有，需要额外申请，那么需要加锁保护
  210.     pthread_mutex_lock(&_mutex);
  211.     if (_pool[index] == NULL) {
  212.         if (_total_mem + index/1024 >= EXTRA_MEM_LIMIT) {
  213.             //当前的开辟的空间已经超过最大限制
  214.             fprintf(stderr, "already use too many memory!\n");
  215.             exit(1);
  216.         }
  218.         io_buf *new_buf = new io_buf(index);
  219.         if (new_buf == NULL) {
  220.             fprintf(stderr, "new io_buf error\n");
  221.             exit(1);
  222.         }
  223.         _total_mem += index/1024;
  224.         pthread_mutex_unlock(&_mutex);
  225.         return new_buf;
  226.     }
  228.     //3 从pool中摘除该内存块
  229.     io_buf *target = _pool[index];
  230.     _pool[index] = target->next;
  231.     pthread_mutex_unlock(&_mutex);
  233.     target->next = NULL;
  235.     return target;
  236. }
  239. //重置一个io_buf,将一个buf 上层不再使用，或者使用完成之后，需要将该buf放回pool中
  240. void buf_pool::revert(io_buf *buffer)
  241. {
  242.     //每个buf的容量都是固定的 在hash的key中取值
  243.     int index = buffer->capacity;
  244.     //重置io_buf中的内置位置指针
  245.     buffer->length = 0;
  246.     buffer->head = 0;
  248.     pthread_mutex_lock(&_mutex);
  249.     //找到对应的hash组 buf首届点地址
  250.     assert(_pool.find(index) != _pool.end());
  252.     //将buffer插回链表头部
  253.     buffer->next = _pool[index];
  254.     _pool[index] = buffer;
  256.     pthread_mutex_unlock(&_mutex);
  257. }
  1. 其中，`buf_pool`构造函数中实现了内存池的hash预开辟内存工作，具体的数据结构如下
  1. //buf_pool -->  [m4K] --> io_buf-io_buf-io_buf-io_buf...
  2. //              [m16K] --> io_buf-io_buf-io_buf-io_buf...
  3. //              [m64K] --> io_buf-io_buf-io_buf-io_buf...
  4. //              [m256K] --> io_buf-io_buf-io_buf-io_buf...
  5. //              [m1M] --> io_buf-io_buf-io_buf-io_buf...
  6. //              [m4M] --> io_buf-io_buf-io_buf-io_buf...
  7. //              [m8M] --> io_buf-io_buf-io_buf-io_buf...
  1. `alloc_buf()`方法，是调用者从内存池中取出一块内存，如果最匹配的内存块存在，则返回，并将该块内存从buf_pool中摘除掉，如果没有则开辟一个内存出来。     `revert()`方法则是将已经使用完的`io_buf`重新放回`buf_pool`中。

3.3 读写buffer机制

  1.     那么接下来我们就需要实现一个专门用来读(输入)数据的`input_buf`和专门用来写(输出)数据的`output_buf`类了。由于这两个人都应该拥有一些`io_buf`的特性，所以我们先定义一个基础的父类`reactor_buf`。

A. reactor_buf类

lars_reactor/include/reactor_buf.h

  1. #pragma once
  2. #include "io_buf.h"
  3. #include "buf_pool.h"
  4. #include <assert.h>
  5. #include <unistd.h>
  8. /*
  9.  * 给业务层提供的最后tcp_buffer结构
  10.  * */
  11. class reactor_buf {
  12. public:
  13.     reactor_buf();
  14.     ~reactor_buf();
  16.     const int length() const;
  17.     void pop(int len);
  18.     void clear();
  20. protected:
  21.     io_buf *_buf;
  22. };
  1.     这个的作用就是将io_buf作为自己的一个成员，然后做了一些包装。具体方法实现如下。

lars_reactor/src/reactor.cpp

  1. #include "reactor_buf.h"
  2. #include <sys/ioctl.h>
  3. #include <string.h>
  5. reactor_buf::reactor_buf() 
  6. {
  7.     _buf = NULL;
  8. }
  10. reactor_buf::~reactor_buf()
  11. {
  12.     clear();
  13. }
  15. const int reactor_buf::length() const 
  16. {
  17.     return _buf != NULL? _buf->length : 0;
  18. }
  20. void reactor_buf::pop(int len) 
  21. {
  22.     assert(_buf != NULL && len <= _buf->length);
  24.     _buf->pop(len);
  26.     //当此时_buf的可用长度已经为0
  27.     if(_buf->length == 0) {
  28.         //将_buf重新放回buf_pool中
  29.         buf_pool::instance()->revert(_buf);
  30.         _buf = NULL;
  31.     }
  32. }
  34. void reactor_buf::clear()
  35. {
  36.     if (_buf != NULL)  {
  37.         //将_buf重新放回buf_pool中
  38.         buf_pool::instance()->revert(_buf);
  39.         _buf = NULL;
  40.     }
  41. }

B. input_buf类

  1. 接下来就可以集成`reactor_buf`类实现`input_buf`类的设计了。

lars_reactor/include/reactor_buf.h

  1. //读(输入) 缓存buffer
  2. class input_buf : public reactor_buf 
  3. {
  4. public:
  5.     //从一个fd中读取数据到reactor_buf中
  6.     int read_data(int fd);
  8.     //取出读到的数据
  9.     const char *data() const;
  11.     //重置缓冲区
  12.     void adjust();
  13. };
  1.     其中data()方法即取出已经读取的数据，adjust()含义和`io_buf`含义一致。主要是`read_data()`方法。具体实现如下。

lars_reactor/src/reactor.cpp

  1. //从一个fd中读取数据到reactor_buf中
  2. int input_buf::read_data(int fd)
  3. {
  4.     int need_read;//硬件有多少数据可以读
  6.     //一次性读出所有的数据
  7.     //需要给fd设置FIONREAD,
  8.     //得到read缓冲中有多少数据是可以读取的
  9.     if (ioctl(fd, FIONREAD, &need_read) == -1) {
  10.         fprintf(stderr, "ioctl FIONREAD\n");
  11.         return -1;
  12.     }
  15.     if (_buf == NULL) {
  16.         //如果io_buf为空,从内存池申请
  17.         _buf = buf_pool::instance()->alloc_buf(need_read);
  18.         if (_buf == NULL) {
  19.             fprintf(stderr, "no idle buf for alloc\n");
  20.             return -1;
  21.         }
  22.     }
  23.     else {
  24.         //如果io_buf可用，判断是否够存
  25.         assert(_buf->head == 0);
  26.         if (_buf->capacity - _buf->length < (int)need_read) {
  27.             //不够存，冲内存池申请
  28.             io_buf *new_buf = buf_pool::instance()->alloc_buf(need_read+_buf->length);
  29.             if (new_buf == NULL) {
  30.                 fprintf(stderr, "no ilde buf for alloc\n");
  31.                 return -1;
  32.             }
  33.             //将之前的_buf的数据考到新申请的buf中
  34.             new_buf->copy(_buf);
  35.             //将之前的_buf放回内存池中
  36.             buf_pool::instance()->revert(_buf);
  37.             //新申请的buf成为当前io_buf
  38.             _buf = new_buf;
  39.         }
  40.     }
  42.     //读取数据
  43.     int already_read = 0;
  44.     do { 
  45.         //读取的数据拼接到之前的数据之后
  46.         if(need_read == 0) {
  47.             //可能是read阻塞读数据的模式，对方未写数据
  48.             already_read = read(fd, _buf->data + _buf->length, m4K);
  49.         } else {
  50.             already_read = read(fd, _buf->data + _buf->length, need_read);
  51.         }
  52.     } while (already_read == -1 && errno == EINTR); //systemCall引起的中断 继续读取
  53.     if (already_read > 0)  {
  54.         if (need_read != 0) {
  55.             assert(already_read == need_read);
  56.         }
  57.         _buf->length += already_read;
  58.     }
  60.     return already_read;
  61. }
  63. //取出读到的数据
  64. const char *input_buf::data() const 
  65. {
  66.     return _buf != NULL ? _buf->data + _buf->head : NULL;
  67. }
  69. //重置缓冲区
  70. void input_buf::adjust()
  71. {
  72.     if (_buf != NULL) {
  73.         _buf->adjust();
  74.     }
  75. }

C. output_buf类

  1. 接下来就可以集成`reactor_buf`类实现`output_buf`类的设计了。

lars_reactor/include/reactor_buf.h

  1. //写(输出)  缓存buffer
  2. class output_buf : public reactor_buf 
  3. {
  4. public:
  5.     //将一段数据 写到一个reactor_buf中
  6.     int send_data(const char *data, int datalen);
  8.     //将reactor_buf中的数据写到一个fd中
  9.     int write2fd(int fd);
  10. };
  1. `send_data()`方法主要是将数据写到`io_buf`中，实际上并没有做真正的写操作。而是当调用`write2fd`方法时，才会将`io_buf`的数据写到对应的fd中。send_data是做一些buf内存块的申请等工作。具体实现如下

lars_reactor/src/reactor.cpp

  1. //将一段数据 写到一个reactor_buf中
  2. int output_buf::send_data(const char *data, int datalen)
  3. {
  4.     if (_buf == NULL) {
  5.         //如果io_buf为空,从内存池申请
  6.         _buf = buf_pool::instance()->alloc_buf(datalen);
  7.         if (_buf == NULL) {
  8.             fprintf(stderr, "no idle buf for alloc\n");
  9.             return -1;
  10.         }
  11.     }
  12.     else {
  13.         //如果io_buf可用，判断是否够存
  14.         assert(_buf->head == 0);
  15.         if (_buf->capacity - _buf->length < datalen) {
  16.             //不够存，冲内存池申请
  17.             io_buf *new_buf = buf_pool::instance()->alloc_buf(datalen+_buf->length);
  18.             if (new_buf == NULL) {
  19.                 fprintf(stderr, "no ilde buf for alloc\n");
  20.                 return -1;
  21.             }
  22.             //将之前的_buf的数据考到新申请的buf中
  23.             new_buf->copy(_buf);
  24.             //将之前的_buf放回内存池中
  25.             buf_pool::instance()->revert(_buf);
  26.             //新申请的buf成为当前io_buf
  27.             _buf = new_buf;
  28.         }
  29.     }
  31.     //将data数据拷贝到io_buf中,拼接到后面
  32.     memcpy(_buf->data + _buf->length, data, datalen);
  33.     _buf->length += datalen;
  35.     return 0;
  36. }
  38. //将reactor_buf中的数据写到一个fd中
  39. int output_buf::write2fd(int fd)
  40. {
  41.     assert(_buf != NULL && _buf->head == 0);
  43.     int already_write = 0;
  45.     do { 
  46.         already_write = write(fd, _buf->data, _buf->length);
  47.     } while (already_write == -1 && errno == EINTR); //systemCall引起的中断，继续写
  50.     if (already_write > 0) {
  51.         //已经处理的数据清空
  52.         _buf->pop(already_write);
  53.         //未处理数据前置，覆盖老数据
  54.         _buf->adjust();
  55.     }
  57.     //如果fd非阻塞，可能会得到EAGAIN错误
  58.     if (already_write == -1 && errno == EAGAIN) {
  59.         already_write = 0;//不是错误，仅仅返回0，表示目前是不可以继续写的
  60.     }
  62.     return already_write;
  63. }
  1.     现在我们已经完成了内存管理及读写buf机制的实现，接下来就要简单的测试一下，用我们之前的V0.1版本的reactor server来测试。

3.4 完成Lars Reactor V0.2开发

A. 修改tcp_server

  1.     主要修正do_accept()方法，加上reactor_buf机制.

lars_reactor/src/tcp_server.cpp

  1. #include <stdio.h>
  2. #include <stdlib.h>
  3. #include <string.h>
  4. #include <strings.h>
  6. #include <unistd.h>
  7. #include <signal.h>
  8. #include <sys/types.h>          /* See NOTES */
  9. #include <sys/socket.h>
  10. #include <arpa/inet.h>
  11. #include <errno.h>
  13. #include "tcp_server.h"
  14. #include "reactor_buf.h"
  18. //server的构造函数
  19. tcp_server::tcp_server(const char *ip, uint16_t port)
  20. {
  21.     //...
  22. }
  24. //开始提供创建链接服务
  25. void tcp_server::do_accept()
  26. {
  27.     int connfd;    
  28.     while(true) {
  29.         //accept与客户端创建链接
  30.         printf("begin accept\n");
  31.         connfd = accept(_sockfd, (struct sockaddr*)&_connaddr, &_addrlen);
  32.         if (connfd == -1) {
  33.             if (errno == EINTR) {
  34.                 fprintf(stderr, "accept errno=EINTR\n");
  35.                 continue;
  36.             }
  37.             else if (errno == EMFILE) {
  38.                 //建立链接过多，资源不够
  39.                 fprintf(stderr, "accept errno=EMFILE\n");
  40.             }
  41.             else if (errno == EAGAIN) {
  42.                 fprintf(stderr, "accept errno=EAGAIN\n");
  43.                 break;
  44.             }
  45.             else {
  46.                 fprintf(stderr, "accept error");
  47.                 exit(1);
  48.             }
  49.         }
  50.         else {
  51.             //accept succ!
  53.             int ret = 0;
  54.             input_buf ibuf;
  55.             output_buf obuf;
  57.             char *msg = NULL;
  58.             int msg_len = 0;
  59.             do { 
  60.                 ret = ibuf.read_data(connfd);
  61.                 if (ret == -1) {
  62.                     fprintf(stderr, "ibuf read_data error\n");
  63.                     break;
  64.                 }
  65.                 printf("ibuf.length() = %d\n", ibuf.length());
  68.                 //将读到的数据放在msg中
  69.                 msg_len = ibuf.length();
  70.                 msg = (char*)malloc(msg_len);
  71.                 bzero(msg, msg_len);
  72.                 memcpy(msg, ibuf.data(), msg_len);
  73.                 ibuf.pop(msg_len);
  74.                 ibuf.adjust();
  76.                 printf("recv data = %s\n", msg);
  78.                 //回显数据
  79.                 obuf.send_data(msg, msg_len);
  80.                 while(obuf.length()) {
  81.                     int write_ret = obuf.write2fd(connfd);
  82.                     if (write_ret == -1) {
  83.                         fprintf(stderr, "write connfd error\n");
  84.                         return;
  85.                     }
  86.                     else if(write_ret == 0) {
  87.                         //不是错误，表示此时不可写
  88.                         break;
  89.                     }
  90.                 }
  93.                 free(msg);
  95.             } while (ret != 0);     
  98.             //Peer is closed
  99.             close(connfd);
  100.         }
  101.     }
  102. }

编译生成新的liblreactor.a

  1. $cd lars_reactor/
  2. $make
  3. g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/tcp_server.o src/tcp_server.cpp -I./include
  4. g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/io_buf.o src/io_buf.cpp -I./include
  5. g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/reactor_buf.o src/reactor_buf.cpp -I./include
  6. g++ -g -O2 -Wall -fPIC -Wno-deprecated -c -o src/buf_pool.o src/buf_pool.cpp -I./include
  7. mkdir -p lib
  8. 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

  1.     我们将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

  1. CXX=g++
  2. CFLAGS=-g -O2 -Wall -fPIC -Wno-deprecated 
  4. INC=-I../../include
  5. LIB=-L../../lib -llreactor -lpthread
  6. OBJS = $(addsuffix .o, $(basename $(wildcard *.cc)))
  8. all:
  9.     $(CXX) -o lars_reactor $(CFLAGS)  lars_reactor.cpp $(INC) $(LIB)
  11. clean:
  12.     -rm -f *.o lars_reactor

编译在lars_reactor/example/lars_reactor_0.2/

  1. $ cd lars_reactor/example/lars_reactor_0.2/
  2. $ make
  3. g++ -o lars_reactor -g -O2 -Wall -fPIC -Wno-deprecated   lars_reactor.cpp -I../../include -L../../lib -llreactor  -lpthread

C. 测试

启动server

  1. $ ./lars_reactor 
  2. begin accept

启动client

  1. $ nc 127.0.0.1 7777

客户端输入 文字，效果如下：

服务端：

  1. ibuf.length() = 21
  2. recv data = hello lars, By Aceld

客户端：

  1. $ nc 127.0.0.1 7777
  2. hello lars, By Aceld
  3. hello lars, By Aceld
  1. ok!现在我们的读写buffer机制已经成功的集成到我们的lars网络框架中了。