本章将带你学习Linux下最高效的异步IO事件处理函数epoll的使用。

6-1 epoll基本知识

图片.png
图片.png
图片.png
图片.png
图片.png
图片.png
图片.png

6-2 epoll高性能服务器的实现

tcp_server_epoll.c

  1. #include <stdio.h>
  2. #include <unistd.h>
  3. #include <stdlib.h>
  4. #include <string.h>
  5. #include <fcntl.h>
  6. #include <errno.h>
  8. #include <sys/epoll.h>
  9. #include <sys/types.h>
  10. #include <sys/socket.h>
  11. #include <netinet/in.h>
  13. #define PORT 8888
  15. #define FD_SIZE 20
  16. #define MAX_EVENTS 20
  17. #define TIME_OUT 500
  19. #define MESSAGE_SIZE 1024
  22. int main(){
  24.   int ret = -1;
  26.   int socket_fd = -1;
  27.   int accept_fd = -1;
  29.   int flags = 1;
  30.   int backlog = 10;
  32.   struct sockaddr_in local_addr,remote_addr;
  34.   struct epoll_event ev, events[FD_SIZE];
  35.   int epoll_fd = -1; 
  36.   int event_number = 0;
  38.   //creat a tcp socket
  39.   socket_fd = socket(AF_INET, SOCK_STREAM, 0);
  40.   if ( socket_fd  == -1 ){
  41.     perror("create socket error");
  42.     exit(1);
  43.   }
  45.   //set REUSERADDR
  46.   ret = setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, (char *)&flags, sizeof(flags)); 
  47.   if ( ret == -1 ){
  48.     perror("setsockopt error");
  49.   }
  51.   //set NONBLOCK
  52.   flags = fcntl(socket_fd, F_GETFL, 0);
  53.   fcntl(socket_fd, F_SETFL, flags|O_NONBLOCK);
  55.   //set address
  56.   local_addr.sin_family = AF_INET;
  57.   local_addr.sin_port = htons(PORT);
  58.   local_addr.sin_addr.s_addr = INADDR_ANY;
  59.   bzero(&(local_addr.sin_zero),8);
  61.   //bind addr
  62.   ret = bind(socket_fd, (struct sockaddr *)&local_addr, sizeof(struct sockaddr_in));
  63.   if( ret == -1 ) {
  64.     perror("bind error");
  65.     exit(1);
  66.   }
  68.   if (listen(socket_fd, backlog) == -1 ){
  69.     perror("listen error");
  70.     exit(1);
  71.   }
  73.   //create epoll
  74.   epoll_fd = epoll_create(256);//the size argument is ignored
  75.   ev.data.fd=socket_fd;
  76.   ev.events=EPOLLIN;
  77.   epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &ev); //将socket_fd 添加到epoll中
  78.   for(;;){
  79.     //events 表示一共有多少事件被侦听
  80.     //MAX_EVENTS 表示在events个事件中，本次调用最多能返回多少个被解发的事件
  81.     //TIME_OUT 表示本次调用最多等多长时间
  82.     //event_number 表示本次调用真正有多少事件被解发
  83.     event_number = epoll_wait(epoll_fd, events, MAX_EVENTS, TIME_OUT);
  84.     for(int i=0; i < event_number; i++){
  86.       if(events[i].data.fd == socket_fd){ // 如果是侦听端口的事件
  88.         printf("listen event... \n");
  90.         int addr_len = sizeof( struct sockaddr_in );
  91.         accept_fd = accept(socket_fd, (struct sockaddr *)&remote_addr, &addr_len);
  93.         //将新创建的socket设置为 NONBLOCK 模式
  94.         flags = fcntl(accept_fd, F_GETFL, 0);
  95.         fcntl(accept_fd, F_SETFL, flags|O_NONBLOCK);
  97.         ev.data.fd=accept_fd;
  98.         ev.events=EPOLLIN | EPOLLET;
  99.         epoll_ctl(epoll_fd, EPOLL_CTL_ADD, accept_fd, &ev);
  101.         printf("new accept fd:%d\n",accept_fd);
  103.       } else if(events[i].events & EPOLLIN){
  105.         //printf("accept event :%d\n",i);
  107.         char in_buf[MESSAGE_SIZE];
  108.         memset(in_buf, 0, MESSAGE_SIZE);
  110.         //receive data
  111.         ret = recv( events[i].data.fd, &in_buf, MESSAGE_SIZE, 0 );
  112.         if(ret == MESSAGE_SIZE ){
  113.           printf("maybe have data....");
  114.         }
  116.         if(ret <= 0){
  118.             switch (errno){
  119.               case EAGAIN: //说明暂时已经没有数据了，要等通知
  120.                 break;
  121.               case EINTR: //被终断了，再来一次
  122.                 printf("recv EINTR... \n");
  123.                 ret = recv(events[i].data.fd, &in_buf, MESSAGE_SIZE, 0);
  124.                 break;
  125.               default:
  126.                 printf("the client is closed, fd:%d\n", events[i].data.fd);
  127.                 epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, &ev); 
  128.                 close(events[i].data.fd);
  129.                 ;
  130.             }
  131.         }
  133.         printf(">>>receive message:%s\n", in_buf);
  134.         send(events[i].data.fd, &in_buf, ret, 0);
  135.       }
  136.     }
  137.   }
  138.   return 0;
  139. }

6-3 epoll+fork进行性能优化

tcp_server_epoll_fork.c

  1. #include <stdio.h>
  2. #include <unistd.h>
  3. #include <stdlib.h>
  4. #include <string.h>
  5. #include <fcntl.h>
  6. #include <errno.h>
  8. #include <sys/epoll.h>
  9. #include <sys/types.h>
  10. #include <sys/wait.h>
  11. #include <sys/socket.h>
  12. #include <netinet/in.h>
  14. #define PORT 8111
  16. #define FD_SIZE 20
  17. #define MAX_EVENTS 20
  18. #define TIME_OUT 500
  20. #define MESSAGE_SIZE 1024
  22. //一般是系统CPU核数*2 +1 
  23. #define NB_PROCESS 4
  26. int main(){
  28.   int ret = -1;
  30.   int socket_fd = -1;
  31.   int accept_fd = -1;
  33.   int flags = 1;
  34.   int backlog = 10;
  36.   struct sockaddr_in local_addr,remote_addr;
  38.   struct epoll_event ev, events[FD_SIZE];
  39.   int epoll_fd = -1; 
  40.   int event_number = 0;
  42.   int pid;
  43.   int status;
  44.   int max_subprocess = NB_PROCESS;
  46.   //creat a tcp socket
  47.   socket_fd = socket(AF_INET, SOCK_STREAM, 0);
  48.   if ( socket_fd  == -1 ){
  49.     perror("create socket error");
  50.     exit(1);
  51.   }
  53.   //set REUSERADDR
  54.   ret = setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, (char *)&flags, sizeof(flags)); 
  55.   if ( ret == -1 ){
  56.     perror("setsockopt error");
  57.   }
  59.   //set NONBLOCK
  60.   flags = fcntl(socket_fd, F_GETFL, 0);
  61.   fcntl(socket_fd, F_SETFL, flags|O_NONBLOCK);
  63.   //set address
  64.   local_addr.sin_family = AF_INET;
  65.   local_addr.sin_port = htons(PORT);
  66.   local_addr.sin_addr.s_addr = INADDR_ANY;
  67.   bzero(&(local_addr.sin_zero),8);
  69.   //bind addr
  70.   ret = bind(socket_fd, (struct sockaddr *)&local_addr, sizeof(struct sockaddr_in));
  71.   if( ret == -1 ) {
  72.     perror("bind error");
  73.     exit(1);
  74.   }
  76.   if (listen(socket_fd, backlog) == -1 ){
  77.     perror("listen error");
  78.     exit(1);
  79.   }
  81.   //fork some subprocess
  82.   for(int a=0; a < max_subprocess; a++){
  83.     if(pid !=0){
  84.       pid = fork(); 
  85.     }
  86.   }
  88.   //child process
  89.   if(pid == 0) {
  91.     printf("create an new child process...");
  92.     //create epoll
  93.     epoll_fd = epoll_create(256);//the size argument is ignored
  94.     ev.data.fd=socket_fd;
  95.     ev.events=EPOLLIN;
  96.     epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &ev); //将socket_fd 添加到epoll中
  97.     for(;;){
  98.       //events 表示一共有多少事件被侦听
  99.       //MAX_EVENTS 表示在events个事件中，本次调用最多能返回多少个被解发的事件
  100.       //TIME_OUT 表示本次调用最多等多长时间
  101.       //event_number 表示本次调用真正有多少事件被解发
  102.       event_number = epoll_wait(epoll_fd, events, MAX_EVENTS, TIME_OUT);
  103.       for(int i=0; i < event_number; i++){
  105.         if(events[i].data.fd == socket_fd){ // 如果是侦听端口的事件
  107.           printf("listen event... \n");
  109.           int addr_len = sizeof( struct sockaddr_in );
  110.           accept_fd = accept(socket_fd, (struct sockaddr *)&remote_addr, &addr_len);
  112.           //将新创建的socket设置为 NONBLOCK 模式
  113.           flags = fcntl(accept_fd, F_GETFL, 0);
  114.           fcntl(accept_fd, F_SETFL, flags|O_NONBLOCK);
  116.           ev.data.fd=accept_fd;
  117.           ev.events=EPOLLIN | EPOLLET;
  118.           epoll_ctl(epoll_fd, EPOLL_CTL_ADD, accept_fd, &ev);
  120.           printf("new accept fd:%d\n",accept_fd);
  122.         } else if(events[i].events & EPOLLIN){
  124.           //printf("accept event :%d\n",i);
  126.           char in_buf[MESSAGE_SIZE];
  127.           memset(in_buf, 0, MESSAGE_SIZE);
  129.           //receive data
  130.           ret = recv( events[i].data.fd, &in_buf, MESSAGE_SIZE, 0 );
  131.           if(ret == MESSAGE_SIZE ){
  132.             printf("maybe have data....");
  133.           }
  135.           if(ret <= 0){
  137.             switch (errno){
  138.               case EAGAIN:
  139.                 ret = recv(events[i].data.fd, &in_buf, MESSAGE_SIZE, 0);
  140.                 break;
  141.               case EINTR:
  142.                 printf("recv EINTR... \n");
  143.                 break;
  144.               default:
  145.                 printf("the client is closed, fd:%d\n", events[i].data.fd);
  146.                 epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, &ev); 
  147.                 close(events[i].data.fd);
  148.                 ;
  149.             }
  150.           }
  152.           printf(">>>receive message:%s\n", in_buf);
  153.           send(events[i].data.fd, &in_buf, ret, 0);
  154.         }
  155.       }
  156.     }
  158.   }else {// pid == 0
  159.     //wait child process to quit 
  160.     wait(&status);
  161.   }
  163.   return 0;
  164. }

epoll在非阻塞的情况下，还是会有惊群现象的。每当有信息过来，总会有一些进程被惊起，抢占消息来处理，其他的进程就做无用功。