IO多路复用

IO多路复用—epoll

epoll可以理解为event poll，不同于忙轮询和无差别轮询，epoll只会把哪个流发生了怎样的I/O事件通知我们。此时我们对这些流的操作都是有意义的。复杂度降低到了O(k)，k为产生I/O事件的流的个数。

epoll的相关操作：

epoll_create 创建一个epoll句柄，一般epfd = epoll_create(int size);

　　　　size用来告诉内核这个监听的数量最大值。注意！不是fd的最大值，是fd的个数最大值。

　　　　当创建好epoll句柄后，它就会占用一个fd值，所以在使用完epoll后，必须调用close()，否则可能导致fd被耗尽。

epoll_ctl (int epfd, int op, int fd, struct epoll_event *event)，往epoll对象中增加/删除某一个流的某一个事件

　　　　第一个参数是epoll_create()的返回值，第二个参数表示动作，用三个宏来表示：

　　　　　　　　EPOLL_CTL_ADD：注册新的fd到epfd中；

　　　　　　　　EPOLL_CTL_MOD：修改已经注册的fd的监听事件；

　　　　　　　　EPOLL_CTL_DEL：从epfd中删除一个fd；

　　　　第三个参数是需要监听的fd，第四个参数是告诉内核需要监听什么事件，struct epoll_event结构如下：

　　　　　　　　struct epoll_event {

　　　　　　　　　　__uint32_t events; /* Epoll events */

　　　　　　　　　　epoll_data_t data; /* User data variable */

　　　　　　　　};

　　　　　　　　typedef union epoll_data{

　　　　　　　　　　void *ptr;

　　　　　　　　　　int fd;

　　　　　　　　　　__uint32_t u32;

　　　　　　　　　　__uint64_t u64;

　　　　　　　　}epoll_data_t;

　　　　　　　　events可以是以下几个宏的集合：

　　　　　　　　　　EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；

　　　　　　　　　　EPOLLOUT：表示对应的文件描述符可以写；

　　　　　　　　　　EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；

　　　　　　　　　　EPOLLERR：表示对应的文件描述符发生错误；

　　　　　　　　　　EPOLLHUP：表示对应的文件描述符被挂断；

　　　　　　　　　　EPOLLET：将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。

　　　　　　　　　　EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再

　　　　　　　　　　　　　　　　　　　次把这个socket加入到EPOLL队列里。

　　　　比如> epoll_ctl(epollfd, EPOLL_CTL_ADD, socket, EPOLLIN);//有缓冲区内有数据时epoll_wait返回
　　　　　　> epoll_ctl(epollfd, EPOLL_CTL_DEL, socket, EPOLLOUT);//有缓冲区可写入时epoll_wait返回

epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout)，等待直到注册事件的产生，返回要处理的事件的数量，并将需处理的套接字集合于参数events内，可以遍历events来处理事件。

　　　　参数events用来从内核得到事件的集合，，参数timeout是超时时间（毫秒，0会立即返回，-1是永久堵塞）。

　　　　参数epfd为epoll句柄

　　　　events为事件集合

　　　　maxevents每次能处理的最大事件数，告之内核这个events有多大，maxevents的值不能大于创建epoll_create()时的size

　　　　参数timeout是超时时间（毫秒，0会马上返回。-1是永久堵塞）。

　　　　返回值：该函数返回须要处理的事件数目。如返回0表示已超时。

利用IO多路复用epoll，建立多客户端连接，客户端向服务器发送需要计算的数和符号，服务器返回计算结果。代码如下：

  //epoll_server.c
  1 #include <stdio.h>
  2 #include <stdlib.h>
  3 #include <string.h>
  4 #include <sys/socket.h>
  5 #include <arpa/inet.h>
  6 #include <sys/epoll.h>
  7 
  8 #define backlog 5
  9 #define OPSZ 4
 10 #define BUF_SIZE 1024
 11 
 12 void error_handling(char *message);
 13 int caculate(int opnd_cnt, int opinfo[], char operator);
 14 
 15 int main()
 16 {    
 17     int result;
 18     int listen_sock, new_sock;
 19     struct sockaddr_in serv_addr, clnt_addr;
 20     
 21     int on = 1, len=sizeof(on);
 22     socklen_t clnt_addr_len = sizeof(clnt_addr);
 23     
 24     listen_sock = socket(PF_INET, SOCK_STREAM, 0);
 25     if(listen_sock == -1)
 26         error_handling("listen_sock() error");
 27     
 28     //设置端口重用
 29     if(setsockopt(listen_sock, SOL_SOCKET, SO_REUSEADDR, (char *)&on, len) < 0)
 30         error_handling("setsockopt() error");
 31     
 32     /*设置服务器地址和端口*/
 33     memset(&serv_addr, 0, sizeof(serv_addr));
 34     serv_addr.sin_family = AF_INET;
 35     serv_addr.sin_port = htons(24241);
 36     serv_addr.sin_addr.s_addr = inet_addr("192.168.1.4");
 37     
 38     //绑定本地地址信息
 39     if(bind(listen_sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1)
 40         error_handling("bind() error");
 41     
 42     if(listen(listen_sock, backlog) == -1)
 43         error_handling("listen() error");
 44     
 45     int epfd, nfds, i;
 46     int maxi = 50, timeout = 5000;
 47     struct epoll_event ev, events[maxi]; //单个事件和事件集合
 48     
 49     epfd = epoll_create(256);
 50     
 51     /*将服务器套接字注册到epfd对象中*/
 52     ev.data.fd = listen_sock;
 53     ev.events = EPOLLIN;
 54     epoll_ctl(epfd, EPOLL_CTL_ADD, listen_sock, &ev);
 55     
 56     char opinfo[BUF_SIZE]; //接收消息的buffer
 57     char operator;  //算术符号( + - * / )
 58     int opnd_cnt;   //被计算的数字个数
 59     int number[opnd_cnt]; //数字数组
 60                         
 61     while(1){
 62         nfds = epoll_wait(epfd, events, maxi, timeout);//等待事件发生
 63         if(nfds == 0){    //没有事件发生，超时
 64             printf("timeout.......
");
 65             continue;
 66         }else{
 67             for(i = 0; i < nfds; i++){
 68                 /*1、监听事件accept*/
 69                 if(events[i].data.fd == listen_sock){
 70                     new_sock = accept(listen_sock, (struct sockaddr*)&clnt_addr, &clnt_addr_len);
 71                     if(new_sock == -1)
 72                         error_handling("accept() error");
 73                     
 74                     /*向epfd对象中注册新建立连接的客户端套接字*/
 75                     ev.data.fd = new_sock;
 76                     ev.events = EPOLLIN;  //设置监听事件为可写
 77                     epoll_ctl(epfd, EPOLL_CTL_ADD, new_sock, &ev);
 78                     
 79                     printf("get a new connet %d.....
", new_sock);
 80                 }
 81                 /*2、可读事件*/
 82                 else if(events[i].events & EPOLLIN){ 
 83                     
 84                     int sockfd = events[i].data.fd;
 85                     
 86                     ssize_t s = read(sockfd, opinfo, sizeof(opinfo)-1);
 87                     if(s == 0){     //客户端发送'
',断开连接
 88                         
 89                         /*删除epfd对象里退出连接的客户端套接字*/
 90                         epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, &events[i]); 
 91                         close(sockfd);
 92                         printf("client %d quit....
", sockfd);
 93                     }else if(s > 0){ 
 94                         int m;
 95                         opnd_cnt = (int)opinfo[0];
 96                      
 97                         for(m = 0; m < opnd_cnt; m++){
 98                             number[m] = (int)opinfo[m*OPSZ+1];
 99                         }
100     
101                         operator = opinfo[m*OPSZ+1];
102                         result = caculate(opnd_cnt, number, operator);
103                          
104                         write(sockfd, &result, 4);  //响应客户端，有可能失败
105                     }
106                 }
107                 /*可写事件*/
108                 else if(events[i].events & EPOLLOUT){ 
109     
110                 }
111             }
112         }
113     }
114     
115     close(listen_sock); 
116     return 0;
117 }
118 
119 int caculate(int opnd_cnt, int number[], char operator)
120 {
121     int value = number[0];
122     switch(operator)
123     {
124         case '+':
125             while(--opnd_cnt)
126                 value += number[opnd_cnt];
127             break;
128         case '-':
129             while(--opnd_cnt)
130                 value -= number[opnd_cnt];
131             break;
132         case '*':
133             while(--opnd_cnt)
134                 value *= number[opnd_cnt];
135             break;
136         case '/':
137             while(--opnd_cnt)
138                 value /= number[opnd_cnt];
139             break;
140     }
141     printf("vaule:%d
", value);
142     return value;
143 }
144 
145 void error_handling(char *message)
146 {
147     printf("%s
", message);
148     exit(1);
149 }

 //epoll_client.c 
 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <string.h>
 4 #include <sys/socket.h>
 5 #include <arpa/inet.h>
 6 
 7 #define backlog 5
 8 #define OPSZ 4
 9 #define BUF_SIZE 1024
10 
11 void error_handling(char *message);
12 
13 int main()
14 {
15     int sock, i;
16     struct sockaddr_in serv_addr;
17     
18     int count, result; 
19     char operator, command[256];
20     char opmsg[BUF_SIZE];
21     
22     sock = socket(PF_INET, SOCK_STREAM, 0);
23     if(sock == -1)
24         error_handling("serv_sock() error");
25     
26     memset(&serv_addr, 0, sizeof(serv_addr));
27     serv_addr.sin_family = AF_INET;
28     serv_addr.sin_port = htons(24241);
29     serv_addr.sin_addr.s_addr = inet_addr("192.168.1.4");
30     
31     if(connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1)
32         error_handling("connect() error");
33     printf("Connect..........
");
34     
35     printf("Continue or quit?(c/q)");
36     while(1){
37         scanf("%s", command);
38         if(strncmp(command, "c", 1) == 0){
39             printf("Operand count:");
40             scanf("%d", &count);
41             opmsg[0] = (char)count;
42                     
43             for(i = 0; i < count; i++)
44             {
45                 printf("Operand %d:", i+1);
46                 scanf("%d", (int *)&opmsg[i*OPSZ+1]);
47             }
48                     
49             printf("Operator:");
50             scanf(" %c", (char *)&opmsg[i*OPSZ+1]);
51                     
52             if(write(sock, opmsg, OPSZ*count+2) == -1)
53                 error_handling("write() error");
54                     
55             read(sock, &result, 4);
56                     
57             printf("result:%d
", result);
58             printf("Continue or quit?(c/q)");
59         }
60         else if(strncmp(command, "q", 1) == 0){    
61             write(sock, '
', 1);
62             break;
63         }
64         else{
65             printf("Continue or quit?(c/q)");
66         }
67     }
68     close(sock);
69     return 0;
70 }
71 
72 void error_handling(char *message)
73 {
74     printf("%s
", message);
75     exit(1);
76 }