4) 事件触发event_loop

接下来我们要尝试添加多路IO的处理机制，当然linux的平台下， 最优的选择就是使用epoll来做，但是用原生的epoll实际上编程起来扩展性不是很强，那么我们就需要封装一套IO事件处理机制。

4.1 io_event基于IO事件封装

我们首先定义一个IO事件类来包括一个时间需要拥有的基本成员信息.

lars_reactor/include/event_base.h

  1. #pragma once
  2. /*
  3.  * 定义一些IO复用机制或者其他异常触发机制的事件封装
  4.  *
  5.  * */
  7. class event_loop;
  9. //IO事件触发的回调函数
  10. typedef void io_callback(event_loop *loop, int fd, void *args);
  12. /*
  13.  * 封装一次IO触发实现 
  14.  * */
  15. struct io_event 
  16. {
  17.              io_event():read_callback(NULL),write_callback(NULL),rcb_args(NULL),wcb_args(NULL) {}
  19.     int mask; //EPOLLIN EPOLLOUT
  20.     io_callback *read_callback; //EPOLLIN事件 触发的回调 
  21.     io_callback *write_callback;//EPOLLOUT事件 触发的回调
  22.     void *rcb_args; //read_callback的回调函数参数
  23.     void *wcb_args; //write_callback的回调函数参数
  24. };
  1.     一个`io_event`对象应该包含 一个epoll的事件标识`EPOLLIN/EPOLLOUT`,和对应事件的处理函数`read_callback`,`write_callback`。他们都应该是`io_callback`类型。然后对应的函数形参。

4.2 event_loop事件循环处理机制

  1.     接下来我们就要通过event_loop类来实现io_event的基本增删操作，放在原生的`epoll`堆中。

lars_reactor/include/event_loop.h

  1. #pragma once
  2. /*
  3.  *
  4.  * event_loop事件处理机制
  5.  *
  6.  * */
  7. #include <sys/epoll.h>
  8. #include <ext/hash_map>
  9. #include <ext/hash_set>
  10. #include "event_base.h"
  12. #define MAXEVENTS 10
  14. // map: fd->io_event 
  15. typedef __gnu_cxx::hash_map<int, io_event> io_event_map;
  16. //定义指向上面map类型的迭代器
  17. typedef __gnu_cxx::hash_map<int, io_event>::iterator io_event_map_it;
  18. //全部正在监听的fd集合
  19. typedef __gnu_cxx::hash_set<int> listen_fd_set;
  21. class event_loop 
  22. {
  23. public:
  24.     //构造，初始化epoll堆
  25.     event_loop();
  27.     //阻塞循环处理事件
  28.     void event_process();
  30.     //添加一个io事件到loop中
  31.     void add_io_event(int fd, io_callback *proc, int mask, void *args=NULL);
  33.     //删除一个io事件从loop中
  34.     void del_io_event(int fd);
  36.     //删除一个io事件的EPOLLIN/EPOLLOUT
  37.     void del_io_event(int fd, int mask);
  39. private:
  40.     int _epfd; //epoll fd
  42.     //当前event_loop 监控的fd和对应事件的关系
  43.     io_event_map _io_evs;
  45.     //当前event_loop 一共哪些fd在监听
  46.     listen_fd_set listen_fds;
  48.     //一次性最大处理的事件
  49.     struct epoll_event _fired_evs[MAXEVENTS];
  51. };

属性:

_epfd:是epoll原生堆的fd。

_io_evs:是一个hash_map对象，主要是方便我们管理fd<—>io_event的对应关系，方便我们来查找和处理。

_listen_fds:记录目前一共有多少个fd正在本我们的event_loop机制所监控.

_fried_evs:已经通过epoll_wait返回的被激活需要上层处理的fd集合.

方法:

event_loop()：构造函数，主要初始化epoll.

event_process():永久阻塞，等待触发的事件，去调用对应的函数callback方法。

add_io_event():绑定一个fd和一个io_event的关系，并添加对应的事件到event_loop中。

del_io_event():从event_loop删除该事件。

  1.     具体实现方法如下：

lars_reactor/src/event_loop.cpp

  1. #include "event_loop.h"
  2. #include <assert.h>
  4. //构造，初始化epoll堆
  5. event_loop::event_loop() 
  6. {
  7.     //flag=0 等价于epll_craete
  8.     _epfd = epoll_create1(0);
  9.     if (_epfd == -1) {
  10.         fprintf(stderr, "epoll_create error\n");
  11.         exit(1);
  12.     }
  13. }
  16. //阻塞循环处理事件
  17. void event_loop::event_process()
  18. {
  19.     while (true) {
  20.         io_event_map_it ev_it;
  22.         int nfds = epoll_wait(_epfd, _fired_evs, MAXEVENTS, 10);
  23.         for (int i = 0; i < nfds; i++) {
  24.             //通过触发的fd找到对应的绑定事件
  25.             ev_it = _io_evs.find(_fired_evs[i].data.fd);
  26.             assert(ev_it != _io_evs.end());
  28.             io_event *ev = &(ev_it->second);
  30.             if (_fired_evs[i].events & EPOLLIN) {
  31.                 //读事件，掉读回调函数
  32.                 void *args = ev->rcb_args;
  33.                 ev->read_callback(this, _fired_evs[i].data.fd, args);
  34.             }
  35.             else if (_fired_evs[i].events & EPOLLOUT) {
  36.                 //写事件，掉写回调函数
  37.                 void *args = ev->wcb_args; 
  38.                 ev->write_callback(this, _fired_evs[i].data.fd, args);
  39.             }
  40.             else if (_fired_evs[i].events &(EPOLLHUP|EPOLLERR)) {
  41.                 //水平触发未处理，可能会出现HUP事件，正常处理读写，没有则清空
  42.                 if (ev->read_callback != NULL) {
  43.                     void *args = ev->rcb_args;
  44.                     ev->read_callback(this, _fired_evs[i].data.fd, args);
  45.                 }
  46.                 else if (ev->write_callback != NULL) {
  47.                     void *args = ev->wcb_args;
  48.                     ev->write_callback(this, _fired_evs[i].data.fd, args);
  49.                 }
  50.                 else {
  51.                     //删除
  52.                     fprintf(stderr, "fd %d get error, delete it from epoll\n", _fired_evs[i].data.fd);
  53.                     this->del_io_event(_fired_evs[i].data.fd);
  54.                 }
  55.             }
  57.         }
  58.     }
  59. }
  61. /*
  62.  * 这里我们处理的事件机制是
  63.  * 如果EPOLLIN 在mask中， EPOLLOUT就不允许在mask中
  64.  * 如果EPOLLOUT 在mask中， EPOLLIN就不允许在mask中
  65.  * 如果想注册EPOLLIN|EPOLLOUT的事件， 那么就调用add_io_event() 方法两次来注册。
  66.  * */
  68. //添加一个io事件到loop中
  69. void event_loop::add_io_event(int fd, io_callback *proc, int mask, void *args)
  70. {
  71.     int final_mask;
  72.     int op;
  74.     //1 找到当前fd是否已经有事件
  75.     io_event_map_it it = _io_evs.find(fd);
  76.     if (it == _io_evs.end()) {
  77.         //2 如果没有操作动作就是ADD
  78.         //没有找到
  79.         final_mask = mask;    
  80.         op = EPOLL_CTL_ADD;
  81.     }
  82.     else {
  83.         //3 如果有操作董酒是MOD
  84.         //添加事件标识位
  85.         final_mask = it->second.mask | mask;
  86.         op = EPOLL_CTL_MOD;
  87.     }
  89.     //4 注册回调函数
  90.     if (mask & EPOLLIN) {
  91.         //读事件回调函数注册
  92.         _io_evs[fd].read_callback = proc;
  93.         _io_evs[fd].rcb_args = args;
  94.     }
  95.     else if (mask & EPOLLOUT) {
  96.         _io_evs[fd].write_callback = proc;
  97.         _io_evs[fd].wcb_args = args;
  98.     }
  100.     //5 epoll_ctl添加到epoll堆里
  101.     _io_evs[fd].mask = final_mask;
  102.     //创建原生epoll事件
  103.     struct epoll_event event;
  104.     event.events = final_mask;
  105.     event.data.fd = fd;
  106.     if (epoll_ctl(_epfd, op, fd, &event) == -1) {
  107.         fprintf(stderr, "epoll ctl %d error\n", fd);
  108.         return;
  109.     }
  111.     //6 将fd添加到监听集合中
  112.     listen_fds.insert(fd);
  113. }
  115. //删除一个io事件从loop中
  116. void event_loop::del_io_event(int fd)
  117. {
  118.     //将事件从_io_evs删除
  119.     _io_evs.erase(fd);
  121.     //将fd从监听集合中删除
  122.     listen_fds.erase(fd);
  124.     //将fd从epoll堆删除
  125.     epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL);
  126. }
  128. //删除一个io事件的EPOLLIN/EPOLLOUT
  129. void event_loop::del_io_event(int fd, int mask)
  130. {
  131.     //如果没有该事件，直接返回
  132.     io_event_map_it it = _io_evs.find(fd);
  133.     if (it == _io_evs.end()) {
  134.         return ;
  135.     }
  137.     int &o_mask = it->second.mask;
  138.     //修正mask
  139.     o_mask = o_mask & (~mask);
  141.     if (o_mask == 0) {
  142.         //如果修正之后 mask为0，则删除
  143.         this->del_io_event(fd);
  144.     }
  145.     else {
  146.         //如果修正之后，mask非0，则修改
  147.         struct epoll_event event;
  148.         event.events = o_mask;
  149.         event.data.fd = fd;
  150.         epoll_ctl(_epfd, EPOLL_CTL_MOD, fd, &event);
  151.     }
  152. }
  1. 这里`del_io_event`提供两个重载，一个是直接删除事件，一个是修正事件。

4.3 Reactor集成event_loop机制

  1.     好了，那么接下来，就让让Lars Reactor框架集成`event_loop`机制。

首先简单修正一个tcp_server.cpp文件，对之前的do_accept()的调度时机做一下修正。

  1. 1. 在`tcp_server`成员新增`event_loop`成员。

lars_reactor/include/tcp_server.h

  1. #pragma once
  3. #include <netinet/in.h>
  4. #include "event_loop.h"
  7. class tcp_server
  8. { 
  9. public: 
  10.     //server的构造函数
  11.     tcp_server(event_loop* loop, const char *ip, uint16_t port); 
  13.     //开始提供创建链接服务
  14.     void do_accept();
  16.     //链接对象释放的析构
  17.     ~tcp_server();
  19. private: 
  20.     int _sockfd; //套接字
  21.     struct sockaddr_in _connaddr; //客户端链接地址
  22.     socklen_t _addrlen; //客户端链接地址长度
  24.       // ============= 新增 ======================
  25.     //event_loop epoll事件机制
  26.     event_loop* _loop;
  27.       // ============= 新增 ======================
  28. };
  1. 构造函数在创建完listen fd之后，添加accept事件。

lars_reactor/src/tcp_server.cpp

  1. //listen fd 客户端有新链接请求过来的回调函数
  2. void accept_callback(event_loop *loop, int fd, void *args)
  3. {
  4.     tcp_server *server = (tcp_server*)args;
  5.     server->do_accept();
  6. }
  8. //server的构造函数
  9. tcp_server::tcp_server(event_loop *loop, const char *ip, uint16_t port)
  10. {
  11.     bzero(&_connaddr, sizeof(_connaddr));
  13.     //忽略一些信号 SIGHUP, SIGPIPE
  14.     //SIGPIPE:如果客户端关闭，服务端再次write就会产生
  15.     //SIGHUP:如果terminal关闭，会给当前进程发送该信号
  16.     if (signal(SIGHUP, SIG_IGN) == SIG_ERR) {
  17.         fprintf(stderr, "signal ignore SIGHUP\n");
  18.     }
  19.     if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) {
  20.         fprintf(stderr, "signal ignore SIGPIPE\n");
  21.     }
  23.     //1. 创建socket
  24.     _sockfd = socket(AF_INET, SOCK_STREAM /*| SOCK_NONBLOCK*/ | SOCK_CLOEXEC, IPPROTO_TCP);
  25.     if (_sockfd == -1) {
  26.         fprintf(stderr, "tcp_server::socket()\n");
  27.         exit(1);
  28.     }
  30.     //2 初始化地址
  31.     struct sockaddr_in server_addr;
  32.     bzero(&server_addr, sizeof(server_addr));
  33.     server_addr.sin_family = AF_INET;
  34.     inet_aton(ip, &server_addr.sin_addr);
  35.     server_addr.sin_port = htons(port);
  37.     //2-1可以多次监听，设置REUSE属性
  38.     int op = 1;
  39.     if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &op, sizeof(op)) < 0) {
  40.         fprintf(stderr, "setsocketopt SO_REUSEADDR\n");
  41.     }
  43.     //3 绑定端口
  44.     if (bind(_sockfd, (const struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
  45.         fprintf(stderr, "bind error\n");
  46.         exit(1);
  47.     }
  49.     //4 监听ip端口
  50.     if (listen(_sockfd, 500) == -1) {
  51.         fprintf(stderr, "listen error\n");
  52.         exit(1);
  53.     }
  55.       // ============= 新增 ======================
  56.     //5 将_sockfd添加到event_loop中
  57.     _loop = loop;
  59.     //6 注册_socket读事件-->accept处理
  60.     _loop->add_io_event(_sockfd, accept_callback, EPOLLIN, this);
  61.       // ============= 新增 ======================
  62. }
  1. 修改do_accept()方法

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"
  16. //临时的收发消息
  17. struct message{
  18.     char data[m4K];
  19.     char len;
  20. };
  21. struct message msg;
  23. void server_rd_callback(event_loop *loop, int fd, void *args);
  24. void server_wt_callback(event_loop *loop, int fd, void *args);
  27. //...省略其他代码
  28. //...省略其他代码
  30. //server read_callback
  31. void server_rd_callback(event_loop *loop, int fd, void *args)
  32. {
  33.     int ret = 0;
  35.     struct message *msg = (struct message*)args;
  36.     input_buf ibuf;
  38.     ret = ibuf.read_data(fd);
  39.     if (ret == -1) {
  40.         fprintf(stderr, "ibuf read_data error\n");
  41.         //删除事件
  42.         loop->del_io_event(fd);
  44.         //对端关闭
  45.         close(fd);
  47.         return;
  48.     }
  49.     if (ret == 0) {
  50.         //删除事件
  51.         loop->del_io_event(fd);
  53.         //对端关闭
  54.         close(fd);
  55.         return ;
  56.     }
  58.     printf("ibuf.length() = %d\n", ibuf.length());
  60.     //将读到的数据放在msg中
  61.     msg->len = ibuf.length();
  62.     bzero(msg->data, msg->len);
  63.     memcpy(msg->data, ibuf.data(), msg->len);
  65.     ibuf.pop(msg->len);
  66.     ibuf.adjust();
  68.     printf("recv data = %s\n", msg->data);
  71.     //删除读事件，添加写事件
  72.     loop->del_io_event(fd, EPOLLIN);
  73.     loop->add_io_event(fd, server_wt_callback, EPOLLOUT, msg);
  74. }
  76. //server write_callback
  77. void server_wt_callback(event_loop *loop, int fd, void *args)
  78. {
  79.     struct message *msg = (struct message*)args;
  80.     output_buf obuf;
  82.     //回显数据
  83.     obuf.send_data(msg->data, msg->len);
  84.     while(obuf.length()) {
  85.         int write_ret = obuf.write2fd(fd);
  86.         if (write_ret == -1) {
  87.             fprintf(stderr, "write connfd error\n");
  88.             return;
  89.         }
  90.         else if(write_ret == 0) {
  91.             //不是错误，表示此时不可写
  92.             break;
  93.         }
  94.     }
  96.     //删除写事件，添加读事件
  97.     loop->del_io_event(fd, EPOLLOUT);
  98.     loop->add_io_event(fd, server_rd_callback, EPOLLIN, msg);
  99. }
  101. //...省略其他代码
  102. //...省略其他代码
  104. //开始提供创建链接服务
  105. void tcp_server::do_accept()
  106. {
  107.     int connfd;    
  108.     while(true) {
  109.         //accept与客户端创建链接
  110.         printf("begin accept\n");
  111.         connfd = accept(_sockfd, (struct sockaddr*)&_connaddr, &_addrlen);
  112.         if (connfd == -1) {
  113.             if (errno == EINTR) {
  114.                 fprintf(stderr, "accept errno=EINTR\n");
  115.                 continue;
  116.             }
  117.             else if (errno == EMFILE) {
  118.                 //建立链接过多，资源不够
  119.                 fprintf(stderr, "accept errno=EMFILE\n");
  120.             }
  121.             else if (errno == EAGAIN) {
  122.                 fprintf(stderr, "accept errno=EAGAIN\n");
  123.                 break;
  124.             }
  125.             else {
  126.                 fprintf(stderr, "accept error");
  127.                 exit(1);
  128.             }
  129.         }
  130.         else {
  131.             //accept succ!
  132.               // ============= 新增 ======================
  133.             this->_loop->add_io_event(connfd, server_rd_callback, EPOLLIN, &msg);
  134.             break;
  135.               // ============= 新增 ======================
  136.         }
  137.     }
  138. }
  140. //...省略其他代码
  141. //...省略其他代码

4.4 完成Lars Reactor V0.3开发

  1.     我们将lars_reactor/example/lars_reactor_0.2的代码复制一份到 lars_reactor/example/lars_reactor_0.3中。

lars_reactor/example/lars_reactor_0.3/lars_reactor.cpp

  1. #include "tcp_server.h"
  3. int main() 
  4. {
  5.     event_loop loop;
  7.     tcp_server server(&loop, "127.0.0.1", 7777);
  9.     loop.event_process();
  11.     return 0;
  12. }

编译。

启动服务器

  1. $ ./lars_reactor

分别启动2个客户端

client1

  1. $ nc 127.0.0.1 7777
  2. hello Iam client1
  3. hello Iam client1  回显

client2

  1. $ nc 127.0.0.1 7777
  2. hello Iam client2
  3. hello Iam client2  回显

服务端打印

  1. $ ./lars_reactor 
  2. begin accept
  3. ibuf.length() = 18
  4. recv data = hello Iam client1
  6. begin accept
  7. ibuf.length() = 18
  8. recv data = hello Iam client2

目前我们已经成功将event_loop机制加入到reactor中了，接下来继续添加功能。