epoll example - ET - LT

LT模式：epoll就是一个快速版poll，可读可写就绪条件和传统poll一致
ET模式：为了避免Starvation，建议 
         1）文件描述符设置为非阻塞 
         2）只在read或write返回EAGAIN后，才能调用下一次epoll_wait 
         3）应用层维护一个就绪链表，进行轮询，可以防止大量IO时在一个描述符上长期read或write（因为只有等到read或 write返回EAGAIN后才               表示该描述符处理完毕）而令其它描述符starve
      理解ET的含义后，上面那些操作其实都是显然的。以wirte为例说明，LT时只要有一定范围的空闲写缓存区，每次epoll_wait都是可写条件就 绪，但是ET时从第一次可写就绪后，epoll_wait不再得到该描述符可写就绪通知直到程序使描述符变为非可写就绪（比如write收到 EAGAIN）后，epoll_wait才可能继续收到可写就绪通知（比如有空闲可写缓存）
      其实ET相对于LT来说，把文件描述符状态跟踪的部分责任由内核空间推到用户空间，内核只关心状态切换即从未就绪到就绪切换时才通知用户，至于保持就绪 状态的，内核不再通知用户，这样在实现非阻塞模型时更方便，不需要每次操作都先查看文件描述符状态。

使用linux epoll模型，水平触发模式（Level-Triggered）, 当socket可写时，会不停的触发socket可写的事件，如何处理？

1. 当需要向socket写数据时，将该socket加入到epoll模型(epoll_ctl), 等待可写事件。接收到socket可写事件后，调用write()或send()发送数据。当数据全部写完后， 将socket描述符移出epoll模型。

这种方式的缺点是： 即使发送很少的数据，也要将socket加入、移出epoll模型。有一定的操作代价。

2. 向socket写数据时，不将socket加入到epoll模型；而是直接调用send()发送; 只有当或send()返回错误码EAGAIN（系统缓存满），才将socket加入到epoll模型，等待可写事件后，再发送数据。

全部数据发送完毕，再移出epoll模型。这种方案的优点： 当用户数据比较少时，不需要epool的事件处理。 

3. 使用Edge-Triggered（边沿触发），这样socket有可写事件，只会触发一次。 可以在应用层做好标记。以避免频繁的调用 epoll_ctl( EPOLL_CTL_ADD, EPOLL_CTL_MOD)。这种方式是epoll的man手册里推荐的方式，性能最高。但如果处理不当容易出错，事件驱动停止。

4. 在epoll_ctl()使用EPOLLONESHOT标志，当事件触发以后，socket会被禁止再次触发。需要再次调用epoll_ctl（EPOLL_CTL_MOD）, 才会接收下一次事件。这种方式可以禁止socket可写事件，应该也会同时禁止可读事件。会带来不便，同时并没有性能优势，因为epoll_ctl有一定的操作代价。

5. 在有些情况下，为了避免针对一个fd读取的字节数量过大而导致影响其他fd的处理，ET模式server并不会读完所以的数据，但是未读完的数据并不会引起fd handler被再此调用，除非有新的数据到来。这时可以先设置fd为可writable，这会使得该fd在下一次epoll_wait时被在此唤醒，此时可以将未读完数据读完。


Server代码

点击(此处)折叠或打开

1. #include
   
2. #include
   
3. 
   
4. #include
   
5. #include
   
6. #include
   
7. 
   
8. #include
   
9. #include
   
10. #include
    
11. #include
    
12. #include
    
13. 
    
14. #include
    
15. #include"stdio.h"
    
16. #include"stdlib.h"
    
17. #include
    
18. #include
    
19. #include
    
20. #include
    
21. #include
    
22. #include
    
23. #include
    
24. #include
    
25. #include
    
26. #include
    
27. #include
    
28. #include
    
29. #include
    
30. #include
    
31. #include
    
32. 
    
33. using namespace std;
    
34. 
    
35. intep;
    
36. struct epoll_event*event;
    
37. intnevent;
    
38. 
    
39.     int
    
40. event_init(intsize)
    
41. {
    
42.     intstatus;
    
43. 
    
44.     ep=epoll_create(size);
    
45.     if(ep<0){
    
46.         return-1;
    
47.     }
    
48.     nevent=size;
    
49. 
    
50.     event=(struct epoll_event*)malloc(sizeof(struct epoll_event)*size);
    
51.     if(event==NULL){
    
52.         status=close(ep);
    
53.         return-1;
    
54.     }
    
55. 
    
56.     memset(event,0,sizeof(struct epoll_event)*size);
    
57.     return 0;
    
58. }
    
59. 
    
60. 
    
61.     int
    
62. event_add_conn(intsd)
    
63. {
    
64.     intstatus;
    
65.     struct epoll_event event;
    
66. 
    
67.     //event.events=(uint32_t)(EPOLLIN|EPOLLOUT|EPOLLET);
    
68.     //event.events=(uint32_t)(EPOLLIN|EPOLLOUT);
    
69.     event.events=(uint32_t)(EPOLLIN);
    
70.     event.data.fd=sd;
    
71. 
    
72.     status=epoll_ctl(ep,EPOLL_CTL_ADD,sd,&event);
    
73.     if(status<0)
    
74.         printf("error: %s\n",strerror(errno));
    
75.     return status;
    
76. }
    
77. 
    
78. 
    
79.     int
    
80. event_del_conn(intsd)
    
81. {
    
82.     intstatus;
    
83. 
    
84.     status=epoll_ctl(ep,EPOLL_CTL_DEL,sd,NULL);
    
85.     if(status<0){
    
86.         cout<<"del error"<
87.     }
    
88. 
    
89.     return status;
    
90. }
    
91. 
    
92. int
    
93. event_add_out(intsd)
    
94. {
    
95.     intstatus;
    
96.     struct epoll_event event;
    
97. 
    
98.     event.events=(uint32_t)(EPOLLIN|EPOLLOUT);
    
99.     //event.events=(uint32_t)(EPOLLIN|EPOLLOUT|EPOLLET);
    
100.     event.data.fd=sd;
     
101. 
     
102.     status=epoll_ctl(ep,EPOLL_CTL_MOD,sd,&event);
     
103.     if(status<0){
     
104.     return-1;
     
105.     }
     
106. 
     
107.     return status;
     
108. }
     
109. 
     
110. int
     
111. event_del_out(intsd)
     
112. {
     
113.     intstatus;
     
114.     struct epoll_event event;
     
115. 
     
116.     event.events=(uint32_t)(EPOLLIN);
     
117.     //event.events=(uint32_t)(EPOLLIN|EPOLLET);
     
118.     event.data.fd=sd;
     
119. 
     
120.     status=epoll_ctl(ep,EPOLL_CTL_MOD,sd,&event);
     
121.     if(status<0){
     
122.     return-1;
     
123.     }
     
124. 
     
125.     return status;
     
126. }
     
127. 
     
128.     int
     
129. event_wait(inttimeout)
     
130. {
     
131.     intnsd;
     
132. 
     
133.     for(;;){
     
134.         nsd=epoll_wait(ep,event,nevent,timeout);
     
135.         if(nsd>0){
     
136.             return nsd;
     
137.         }
     
138. 
     
139.         if(nsd==0){
     
140.             if(timeout==-1){
     
141.                 printf("timeout\n");
     
142.                 return-1;
     
143.             }
     
144. 
     
145.             return 0;
     
146.         }
     
147. 
     
148.         if(errno==EINTR){
     
149.             continue;
     
150.         }
     
151.         return-1;
     
152.     }
     
153. 
     
154.     return 0;
     
155. }
     
156. 
     
157. void setnonblocking(intsock)
     
158. {
     
159.     intopts;
     
160.     opts=fcntl(sock,F_GETFL);
     
161.     if(opts<0)
     
162.     {
     
163.         perror("fcntl(sock,GETFL)");
     
164.         exit(1);
     
165.     }
     
166.     opts=opts|O_NONBLOCK;
     
167.     if(fcntl(sock,F_SETFL,opts)<0)
     
168.     {
     
169.         perror("fcntl(sock,SETFL,opts)");
     
170.         exit(1);
     
171.     }
     
172. }
     
173. 
     
174. intlistenfd;
     
175. #define SERV_PORT 10000
     
176. 
     
177. 
     
178. int
     
179. set_tcpnodelay(intsd)
     
180. {
     
181.     intnodelay;
     
182.     socklen_tlen;
     
183. 
     
184.     nodelay=1;
     
185.     len=sizeof(nodelay);
     
186. 
     
187.     return setsockopt(sd,IPPROTO_TCP,TCP_NODELAY,&nodelay,len);
     
188. }
     
189. 
     
190. 
     
191. intinit_server()
     
192. {
     
193.     struct sockaddr_in serveraddr;
     
194. 
     
195.     listenfd=socket(AF_INET,SOCK_STREAM,0);
     
196.     setnonblocking(listenfd);
     
197.     bzero(&serveraddr,sizeof(serveraddr));
     
198.     serveraddr.sin_family=AF_INET;
     
199.     char*local_addr=(char*)"127.0.0.1";
     
200.     inet_aton(local_addr,&(serveraddr.sin_addr));
     
201.     serveraddr.sin_port=htons(SERV_PORT);
     
202.     bind(listenfd,(sockaddr*)&serveraddr,sizeof(serveraddr));
     
203.     listen(listenfd,1024);    
     
204.     intret=event_add_conn(listenfd);
     
205.     if(ret<0)
     
206.         cout<<"error"<
207.     return 0;
     
208. }
     
209. 
     
210. intprocess_handle()
     
211. {
     
212.     for(;;){
     
213.         intfd=accept(listenfd,NULL,NULL);
     
214.         if(fd<0){
     
215.             if(errno==EINTR){
     
216.                 cout<<"accept on not ready - eintr"<
217.                 continue;
     
218.             }
     
219. 
     
220.             if(errno==EAGAIN||errno==EWOULDBLOCK){
     
221.                 cout<<"accept on not ready - eagain"<
222.                 return 0;
     
223.             }
     
224. 
     
225.             return-1;
     
226.         }
     
227.         elseif(fd>0)
     
228.         {
     
229.             setnonblocking(fd);
     
230.             set_tcpnodelay(fd);
     
231.             intret=event_add_conn(fd);
     
232.             cout<<"accept: "<
233.             if(ret<0)
     
234.                 cout<<"error"<
235.         }
     
236.     }
     
237. }
     
238. 
     
239. intread_data(intfd)
     
240. {
     
241. 
     
242.     for(;;)
     
243.     {
     
244.         char c;
     
245.         intn=read(fd,&c,1);
     
246.         if(n>0)
     
247.             cout<
248.         elseif(n==0)
     
249.         {
     
250.             cout<<"eof"<
251.             event_del_conn(fd);
     
252.             close(fd);
     
253.             return 0;
     
254.         }
     
255.         elseif(n<0)
     
256.         {
     
257.             if(errno==EINTR)
     
258.                 continue;
     
259.             elseif(errno==EAGAIN||errno==EWOULDBLOCK)
     
260.             {
     
261.                 cout<<"again"<
262.                             return 0;    
     
263.             }
     
264.         }
     
265.     }
     
266.     
     
267.     return 0;
     
268. }
     
269. 
     
270. intmain()
     
271. {
     
272.     intret=event_init(100);
     
273.     if(ret<0)
     
274.         cout<<"error"<
275. 
     
276.     init_server();
     
277. 
     
278.     while(1)
     
279.     {
     
280.         inttimeout=-1;
     
281.         intnsd=event_wait(timeout);
     
282.         if(nsd<=0)
     
283.             continue;        
     
284. 
     
285.         inti;    
     
286.         
     
287.         intready=0;
     
288.         for(i=0;i
289.         {
     
290.             struct epoll_event*ev=&event[i];
     
291.             intfd=(int)(ev->data.fd);
     
292.             if(fd==listenfd){
     
293.                 cout<<"listen"<
294.                 process_handle();
     
295.             }
     
296.             else
     
297.             {
     
298.                 uint32_t events=ev->events;
     
299.                 /*errortakes precedence over read|write*/
     
300.                 if(events&EPOLLERR){
     
301.                     printf("event error\n");
     
302.                     break;
     
303.                 }
     
304. 
     
305.                 /*read takes precedence over write*/
     
306.                 if(events&(EPOLLIN|EPOLLHUP)){
     
307.                     printf("event in\n");
     
308.                     read_data(fd);
     
309.                     if(ready==1)
     
310.                     {
     
311.                         event_add_out(fd);
     
312.                     }
     
313.                 }
     
314.             
     
315.                 if(events&(EPOLLOUT|EPOLLHUP))
     
316.                 {
     
317.                     printf("event out");
     
318.                     if(ready==0)
     
319.                     {    
     
320.                         event_del_out(fd);
     
321.                     }
     
322.                     else
     
323.                     {
     
324.                         //send_data(fd);
     
325.                     }
     
326.                     /*the output will be addedinEPOLLIN
     
327.                     *handler*/
     
328.                     break;
     
329.                 }
     
330.             }
     
331.         }
     
332.     }
     
333. 
     
334.     return 0;
     
335. }

Client代码

点击(此处)折叠或打开

1. #include
   
2. #include
   
3. #include
   
4. 
   
5. #include
   
6. #include
   
7. #include
   
8. #include
   
9. #include
   
10. 
    
11. #include
    
12. #include"stdio.h"
    
13. #include"stdlib.h"
    
14. #include
    
15. #include
    
16. #include
    
17. #include
    
18. #include
    
19. #include
    
20. #include
    
21. #include
    
22. #include
    
23. #include
    
24. #include
    
25. #include
    
26. #include
    
27. #include
    
28. #include
    
29. 
    
30. using namespace std;
    
31. 
    
32. #define SERV_PORT 10000
    
33. 
    
34. void setblocking(intsock)
    
35. {
    
36.     intopts;
    
37.     opts=fcntl(sock,F_GETFL);
    
38.     if(opts<0)
    
39.     {
    
40.         perror("fcntl(sock,GETFL)");
    
41.         exit(1);
    
42.     }
    
43.     
    
44.     opts=opts&~O_NONBLOCK;
    
45.     if(fcntl(sock,F_SETFL,opts)<0)
    
46.     {
    
47.         perror("fcntl(sock,SETFL,opts)");
    
48.         exit(1);
    
49.     }
    
50. }
    
51. 
    
52. intmain()
    
53. {
    
54.     struct sockaddr_in serveraddr;
    
55. 
    
56.     intconnfd=socket(AF_INET,SOCK_STREAM,0);
    
57.     bzero(&serveraddr,sizeof(serveraddr));
    
58.     serveraddr.sin_family=AF_INET;
    
59.     char*local_addr=(char*)"127.0.0.1";
    
60.     inet_aton(local_addr,&(serveraddr.sin_addr));
    
61.     serveraddr.sin_port=htons(SERV_PORT);
    
62.     intstatus=connect(connfd,(sockaddr*)&serveraddr,sizeof(serveraddr));
    
63.     if(status!=0){
    
64.         if(errno==EINPROGRESS){
    
65.             return 0;
    
66.         }
    
67.         return-1;
    
68.     }
    
69. 
    
70.     setblocking(connfd);
    
71.     
    
72.     char buffer[1024]={"songweisongweisongwei"};
    
73.     intsize=write(connfd,buffer,strlen(buffer));
    
74.     cout<<"size: "<
75.     
    
76.     sleep(5);
    
77.     
    
78.     size=write(connfd,buffer,strlen(buffer));
    
79.         cout<<"size: "<
80. 
    
81.     return 0;
    
82.