3.2 epoll的反应堆模式实现

epoll反应堆模式的实现-也就是libevent的实现原理。

  1. #include <stdlib.h>
  2. #include <stdio.h> 
  3. #include <stdio.h>
  4. #include <sys/socket.h>
  5. #include <sys/epoll.h>
  6. #include <arpa/inet.h>
  7. #include <fcntl.h>
  8. #include <unistd.h>
  9. #include <errno.h>
  10. #include <string.h>
  11. #include <time.h>
  12. #define MAX_EVENTS  1024
  13. #define BUFLEN 128
  14. #define SERV_PORT   8080
  17. /*
  18.  * status:1表示在监听事件中，0表示不在 
  19.  * last_active:记录最后一次响应时间,做超时处理
  20.  */
  21. struct myevent_s {
  22.     int fd;                 //cfd listenfd
  23.     int events;             //EPOLLIN  EPLLOUT
  24.     void *arg;              //指向自己结构体指针
  25.     void (*call_back)(int fd, int events, void *arg);
  26.     int status;
  27.     char buf[BUFLEN];
  28.     int len;
  29.     long last_active;
  30. };
  32. int g_efd;          /* epoll_create返回的句柄 */
  33. struct myevent_s g_events[MAX_EVENTS+1];   /* +1 最后一个用于 listen fd */
  35. void eventset(struct myevent_s *ev, int fd, void (*call_back)(int, int, void *), void *arg)
  36. {
  37.     ev->fd = fd;
  38.     ev->call_back = call_back;
  39.     ev->events = 0;
  40.     ev->arg = arg;
  41.     ev->status = 0;
  42.     //memset(ev->buf, 0, sizeof(ev->buf));
  43.     //ev->len = 0;
  44.     ev->last_active = time(NULL);
  46.     return;
  47. }
  49. void recvdata(int fd, int events, void *arg);
  50. void senddata(int fd, int events, void *arg);
  52. void eventadd(int efd, int events, struct myevent_s *ev)
  53. {
  54.     struct epoll_event epv = {0, {0}};
  55.     int op;
  56.     epv.data.ptr = ev;
  57.     epv.events = ev->events = events;
  59.     if (ev->status == 1) {
  60.         op = EPOLL_CTL_MOD;
  61.     } 
  62.     else {
  63.         op = EPOLL_CTL_ADD;
  64.         ev->status = 1;
  65.     }
  67.     if (epoll_ctl(efd, op, ev->fd, &epv) < 0)
  68.         printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
  69.     else
  70.         printf("event add OK [fd=%d], op=%d, events[%0X]\n", ev->fd, op, events);
  72.     return;
  73. }
  75. void eventdel(int efd, struct myevent_s *ev)
  76. {
  77.     struct epoll_event epv = {0, {0}};
  79.     if (ev->status != 1)
  80.         return;
  82.     epv.data.ptr = ev;
  83.     ev->status = 0;
  84.     epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv);
  86.     return;
  87. }
  90. void acceptconn(int lfd, int events, void *arg)
  91. {
  92.     struct sockaddr_in cin;
  93.     socklen_t len = sizeof(cin);
  94.     int cfd, i;
  96.     if ((cfd = accept(lfd, (struct sockaddr *)&cin, &len)) == -1) {
  97.         if (errno != EAGAIN && errno != EINTR) {
  98.             /* 暂时不做出错处理 */
  99.         }
  100.         printf("%s: accept, %s\n", __func__, strerror(errno));
  101.         return;
  102.     }
  104.     do {
  105.         for (i = 0; i < MAX_EVENTS; i++) {
  106.             if (g_events[i].status == 0)
  107.                 break;
  108.         }
  110.         if (i == MAX_EVENTS) {
  111.             printf("%s: max connect limit[%d]\n", __func__, MAX_EVENTS);
  112.             break;
  113.         }
  115.         int flag = 0;
  116.         if ((flag = fcntl(cfd, F_SETFL, O_NONBLOCK)) < 0)
  117.         {
  118.             printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
  119.             break;
  120.         }
  122.         eventset(&g_events[i], cfd, recvdata, &g_events[i]);
  123.         eventadd(g_efd, EPOLLIN, &g_events[i]);
  124.     } while(0);
  126.     printf("new connect [%s:%d][time:%ld], pos[%d]\n", inet_ntoa(cin.sin_addr), ntohs(cin.sin_port), g_events[i].last_active, i);
  128.     return;
  129. }
  131. void recvdata(int fd, int events, void *arg)
  132. {
  133.     struct myevent_s *ev = (struct myevent_s *)arg;
  134.     int len;
  136.     len = recv(fd, ev->buf, sizeof(ev->buf), 0);
  137.     eventdel(g_efd, ev);
  139.     if (len > 0) {
  140.         ev->len = len;
  141.         ev->buf[len] = '\0';
  142.         printf("C[%d]:%s\n", fd, ev->buf);
  143.         /* 转换为发送事件 */
  144.         eventset(ev, fd, senddata, ev);
  145.         eventadd(g_efd, EPOLLOUT, ev);
  146.     }
  147.     else if (len == 0) {
  148.         close(ev->fd);
  149.         /* ev-g_events 地址相减得到偏移元素位置 */
  150.         printf("[fd=%d] pos[%d], closed\n", fd, (int)(ev - g_events));
  151.     }
  152.     else {
  153.         close(ev->fd);
  154.         printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
  155.     }
  157.     return;
  158. }
  160. void senddata(int fd, int events, void *arg)
  161. {
  162.     struct myevent_s *ev = (struct myevent_s *)arg;
  163.     int len;
  165.     len = send(fd, ev->buf, ev->len, 0);
  166.     //printf("fd=%d\tev->buf=%s\ttev->len=%d\n", fd, ev->buf, ev->len);
  167.     //printf("send len = %d\n", len);
  169.     eventdel(g_efd, ev);
  170.     if (len > 0) {
  171.         printf("send[fd=%d], [%d]%s\n", fd, len, ev->buf);
  172.         eventset(ev, fd, recvdata, ev);
  173.         eventadd(g_efd, EPOLLIN, ev);
  174.     }
  175.     else {
  176.         close(ev->fd);
  177.         printf("send[fd=%d] error %s\n", fd, strerror(errno));
  178.     }
  180.     return;
  181. }
  183. void initlistensocket(int efd, short port)
  184. {
  185.     int lfd = socket(AF_INET, SOCK_STREAM, 0);
  186.     fcntl(lfd, F_SETFL, O_NONBLOCK);
  187.     eventset(&g_events[MAX_EVENTS], lfd, acceptconn, &g_events[MAX_EVENTS]);
  188.     eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS]);
  190.     struct sockaddr_in sin;
  192.     memset(&sin, 0, sizeof(sin));
  193.     sin.sin_family = AF_INET;
  194.     sin.sin_addr.s_addr = INADDR_ANY;
  195.     sin.sin_port = htons(port);
  197.     bind(lfd, (struct sockaddr *)&sin, sizeof(sin));
  199.     listen(lfd, 20);
  201.     return;
  202. }
  204. int main(int argc, char *argv[])
  205. {
  206.     unsigned short port = SERV_PORT;
  208.     if (argc == 2)
  209.         port = atoi(argv[1]);
  211.     g_efd = epoll_create(MAX_EVENTS+1);
  213.     if (g_efd <= 0)
  214.         printf("create efd in %s err %s\n", __func__, strerror(errno));
  216.     initlistensocket(g_efd, port);
  218.     /* 事件循环 */
  219.     struct epoll_event events[MAX_EVENTS+1];
  221.     printf("server running:port[%d]\n", port);
  222.     int checkpos = 0, i;
  223.     while (1) {
  224.         /* 超时验证，每次测试100个链接，不测试listenfd 当客户端60秒内没有和服务器通信，则关闭此客户端链接 */
  225.         long now = time(NULL);
  226.         for (i = 0; i < 100; i++, checkpos++) {
  227.             if (checkpos == MAX_EVENTS)
  228.                 checkpos = 0;
  229.             if (g_events[checkpos].status != 1)
  230.                 continue;
  231.             long duration = now - g_events[checkpos].last_active;
  232.             if (duration >= 60) {
  233.                 close(g_events[checkpos].fd);
  234.                 printf("[fd=%d] timeout\n", g_events[checkpos].fd);
  235.                 eventdel(g_efd, &g_events[checkpos]);
  236.             }
  237.         }
  238.         /* 等待事件发生 */
  239.         int nfd = epoll_wait(g_efd, events, MAX_EVENTS+1, 1000);
  240.         if (nfd < 0) {
  241.             printf("epoll_wait error, exit\n");
  242.             break;
  243.         }
  244.         for (i = 0; i < nfd; i++) {
  245.             struct myevent_s *ev = (struct myevent_s *)events[i].data.ptr;
  246.             if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
  247.                 ev->call_back(ev->fd, events[i].events, ev->arg);
  248.             }
  249.             if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
  250.                 ev->call_back(ev->fd, events[i].events, ev->arg);
  251.             }
  252.         }
  253.     }
  255.     /* 退出前释放所有资源 */
  256.     return 0;
  257. }