初级epoll问题，希望高手花几分钟解答一下困扰新手几周的问题

xtchina

问题已解决，自己粗心造成

1.为什么用EPOLL_CTL_ADD加入到epoll中的socket会无声无息的消失了？
2.为什么用EPOLL_CTL_DEL删除从epoll_wait中返回的句柄会出现bad file descriptor错误？
3.为什么已经加入到epoll中的sock连接中断，却没有从epoll中得到任何提示？
4.为什么我的socket句柄还没有调用close，系统就把它分配给别的连接用了？

撇开效率不谈，我的代码**问题？网上找了好几周了，实在解决不了，还望高手能够帮忙看看，不胜感激！

1. 
   
2. #include <sys/types.h>
   
3. #include <sys/mman.h>
   
4. #include <sys/resource.h>
   
5. #include <stdio.h>
   
6. #include <errno.h>
   
7. #include <unistd.h>
   
8. #include <string.h>
   
9. #include <fcntl.h>
   
10. #include <unistd.h>
    
11. #include <stdlib.h>
    
12. #include <assert.h>
    
13. 
    
14. #include <pthread.h>
    
15. 
    
16. #include <netinet/tcp.h>
    
17. #include <sys/epoll.h>
    
18. #include <sys/socket.h>
    
19. #include <netinet/in.h>
    
20. #include <arpa/inet.h>
    
21. 
    
22. bool                        woking                                        = true;
    
23. 
    
24. unsigned short        server_port                                = 9001;
    
25. const char*                server_ip                                = "127.0.0.1";
    
26. 
    
27. int                                wait_epevent_num                = 4096;
    
28. int                                ep_client                                = -1;
    
29. int                                ep_server                                = -1;
    
30. int                                buffer_size                                = 4096;
    
31. 
    
32. double                        server_recv_size                = 0.0d;
    
33. double                        server_send_size                = 0.0d;
    
34. 
    
35. double                        client_recv_size                = 0.0d;
    
36. double                        client_send_size                = 0.0d;
    
37. 
    
38. unsigned int        server_inepoll_sock                = 0;
    
39. unsigned int        server_outepoll_sock        = 0;
    
40. 
    
41. unsigned int        client_inepoll_sock                = 0;
    
42. unsigned int        client_outepoll_sock        = 0;
    
43. 
    
44. struct epitem
    
45. {
    
46.         int                                sock;
    
47.         unsigned short        local_port;
    
48.         unsigned short        remote_port;
    
49.         unsigned int        send_data_size;
    
50.         unsigned int        recv_data_size;
    
51. };
    
52. 
    
53. // 获取本端端口
    
54. unsigned short get_sock_local_port(int sock)
    
55. {
    
56.         unsigned short local_port = 0;
    
57.         sockaddr_in addr;
    
58.         socklen_t nl = sizeof(addr);
    
59.         if (0 == getsockname(sock, (sockaddr*)&addr, &nl))
    
60.         {
    
61.                 local_port = ntohs(addr.sin_port);
    
62.         }
    
63. 
    
64.         return local_port;
    
65. }
    
66. // 获取远端端口
    
67. unsigned short get_sock_remote_port(int sock)
    
68. {
    
69.         unsigned short remote_port = 0;
    
70.         sockaddr_in addr;
    
71.         socklen_t nl = sizeof(addr);
    
72.         if (0 == getpeername(sock, (sockaddr*)&addr, &nl))
    
73.         {
    
74.                 remote_port = ntohs(addr.sin_port);
    
75.         }
    
76. 
    
77.         return remote_port;
    
78. }
    
79. 
    
80. // 投入到epoll中
    
81. bool putinto_epoll(int epoll, int sock)
    
82. {
    
83.         if (0 != fcntl(sock, F_SETFL
    
84.                 , fcntl(sock, F_GETFL, 0)|O_NONBLOCK))
    
85.         {
    
86.                 return false;
    
87.         }
    
88. 
    
89.         epitem* newitem = new epitem;
    
90.         if (0 == newitem)
    
91.         {
    
92.                 return false;
    
93.         }
    
94. 
    
95.         newitem->sock = sock;
    
96.         newitem->local_port = get_sock_local_port(sock);
    
97.         newitem->remote_port = get_sock_remote_port(sock);
    
98.         newitem->send_data_size = 0;
    
99.         newitem->recv_data_size = 0;
    
100. 
     
101.         epoll_event eevent;
     
102.         eevent.data.ptr = newitem;
     
103.         eevent.events = EPOLLIN | EPOLLOUT;
     
104. 
     
105.         if (0 != epoll_ctl(epoll, EPOLL_CTL_ADD, sock, &eevent))
     
106.         {       
     
107.                 printf("epoll_ctl call failed [%d:%s]...\r\n", errno, strerror(errno));
     
108.                 delete newitem;
     
109.                 return false;
     
110.         }       
     
111. 
     
112.         // 33%的概率关掉一些刚刚投入到epoll中的sock，仅仅是测试需要
     
113.         if (rand() % 3 == 0)
     
114.         {
     
115.                 close(sock);
     
116.         }
     
117. 
     
118.         return true;
     
119. }
     
120. 
     
121. // 连接线程，不停地进行连接
     
122. void* thread_connect(void* param)
     
123. {
     
124.         printf("begin thread connect...\r\n");
     
125. 
     
126.         sockaddr_in server_addr;
     
127.         memset(&server_addr, 0, sizeof(sockaddr_in));
     
128.         server_addr.sin_family                = AF_INET;
     
129.         server_addr.sin_addr.s_addr = inet_addr(server_ip);
     
130.         server_addr.sin_port                = htons(server_port);
     
131. 
     
132.         while(woking)
     
133.         {
     
134.                 int new_client_sock = -1;               
     
135. 
     
136.                 new_client_sock = socket(PF_INET, SOCK_STREAM, 0);
     
137.                 if (-1        == new_client_sock)
     
138.                 {
     
139.                         sleep(1);
     
140.                         continue;
     
141.                 }
     
142. 
     
143.                 if (0 != connect(new_client_sock, (sockaddr*)&server_addr
     
144.                         , sizeof(sockaddr_in)))
     
145.                 {
     
146.                         goto fail;
     
147.                 }
     
148. 
     
149.                 // 放入epoll队列中
     
150.                 if (!putinto_epoll(ep_client, new_client_sock))
     
151.                 {
     
152.                         goto fail;
     
153.                 }
     
154. 
     
155.                 client_inepoll_sock++;
     
156.                 goto ok;
     
157. 
     
158. fail:
     
159.                 close(new_client_sock);
     
160.                 //sleep(1);
     
161. ok:
     
162.                 ;
     
163.         }
     
164.         printf("end thread connect...\r\n");
     
165.         return 0;
     
166. }
     
167. // 接收线程，不停地接收客户端的连接
     
168. void* thread_accept(void* param)
     
169. {
     
170.         printf("begin thread accept...\r\n");
     
171. 
     
172.         int listen_sock;
     
173.         sockaddr_in server_addr;       
     
174. 
     
175.         listen_sock = socket(PF_INET, SOCK_STREAM, 0);
     
176.         if (-1 == listen_sock)
     
177.         {
     
178.                 printf("end thread accept[%d:%s]...\r\n", errno, strerror(errno));
     
179.                 return 0;
     
180.         }
     
181. 
     
182.         int reuseaddr = 1;
     
183.         if (0 != setsockopt(listen_sock, SOL_SOCKET
     
184.                 , SO_REUSEADDR, &reuseaddr, sizeof(reuseaddr) ))
     
185.         {
     
186.                 printf("end thread accept[%d:%s]...\r\n", errno, strerror(errno));
     
187.                 return 0;
     
188.         }
     
189. 
     
190.         memset(&server_addr, 0, sizeof(sockaddr_in));
     
191.         server_addr.sin_family = AF_INET;
     
192.         server_addr.sin_addr.s_addr = inet_addr(server_ip);
     
193.         server_addr.sin_port = htons(server_port);
     
194. 
     
195.         if (0 != bind(listen_sock, (sockaddr*)&server_addr, sizeof(server_addr)))
     
196.         {
     
197.                 printf("end thread accept[%d:%s]...\r\n", errno, strerror(errno));
     
198.                 return 0;
     
199.         }
     
200. 
     
201.         if (0 != fcntl(listen_sock, F_SETFL
     
202.                 , fcntl(listen_sock, F_GETFL, 0) & ~O_NONBLOCK))
     
203.         {
     
204.                 printf("end thread accept[%d:%s]...\r\n", errno, strerror(errno));
     
205.                 return 0;
     
206.         }
     
207. 
     
208.         if (0 != listen(listen_sock, 32))
     
209.         {
     
210.                 printf("end thread accept[%d:%s]...\r\n", errno, strerror(errno));
     
211.                 return 0;
     
212.         }
     
213. 
     
214.         while(woking)
     
215.         {
     
216.                 int new_accept_sock = -1;
     
217.                 new_accept_sock = accept(listen_sock, 0, 0);
     
218.                 if (-1 == new_accept_sock)
     
219.                 {
     
220.                         goto fail;
     
221.                 }
     
222. 
     
223.                 // 放入epoll队列中
     
224.                 if (!putinto_epoll(ep_server, new_accept_sock))
     
225.                 {
     
226.                         goto fail;
     
227.                 }
     
228. 
     
229.                 server_inepoll_sock++;
     
230.                 goto ok;
     
231. 
     
232. fail:
     
233.                 close(new_accept_sock);
     
234.                 //sleep(1);
     
235. 
     
236. ok:
     
237.                 ;
     
238.         }
     
239. 
     
240.         printf("end thread accept...\r\n");
     
241.         return 0;
     
242. }
     
243. // epoll中的输入处理
     
244. bool do_input(int sock, char* buffer, int buf_size, double& recv_size)
     
245. {
     
246.         assert(buf_size > 0);
     
247.         assert(buffer != 0);
     
248. 
     
249.         for (int i=0; i<8; i++)
     
250.         {
     
251.                 int recvlen = recv(sock, buffer, buf_size, 0);
     
252.                 // return false; 如果这里返回为什么会有double free错误？
     
253.                 recv_size += recvlen>0?recvlen:0;
     
254.                 if (recvlen == 0)
     
255.                 {
     
256.                         return false;
     
257.                 }
     
258.                 else if (recvlen == -1)
     
259.                 {
     
260.                         if (errno == EAGAIN)
     
261.                         {
     
262.                                 return true;
     
263.                         }
     
264.                         if (errno == EINTR)
     
265.                         {
     
266.                                 continue;
     
267.                         }
     
268.                        
     
269.                         return false;
     
270.                 }
     
271.                 else if (recvlen == buf_size)
     
272.                 {
     
273.                         continue;
     
274.                 }
     
275.                 else if (recvlen < buf_size)
     
276.                 {
     
277.                         assert(recvlen > 0);
     
278.                         return true;
     
279.                 }               
     
280.         }
     
281.         // 有太多的数据时，返回false关闭连接
     
282.         return false;
     
283. }
     
284. // epoll中的输出处理
     
285. bool do_output(int sock, char* buffer, int buf_size, double& send_size)
     
286. {
     
287.         for (int i=0; i<4; i++)
     
288.         {
     
289.                 int sendlen = send(sock, buffer, buf_size, 0);
     
290.                 // return false; 如果这里返回为什么会有double free错误？
     
291.                 send_size += sendlen>0?sendlen:0;
     
292.                 if (sendlen == -1)
     
293.                 {
     
294.                         if (errno == EAGAIN)
     
295.                         {
     
296.                                 return true;
     
297.                         }
     
298.                         if (errno == EINTR)
     
299.                         {
     
300.                                 continue;
     
301.                         }
     
302.                         // 为什么有SIGPIPE信号，出错的连接我都关掉了
     
303.                         return false;
     
304.                 }
     
305.                 else if (sendlen == buf_size)
     
306.                 {
     
307.                         continue;
     
308.                 }
     
309.                 else if (sendlen < buf_size)
     
310.                 {
     
311.                         assert(sendlen > 0);
     
312.                         return true;
     
313.                 }               
     
314.         }
     
315. 
     
316.         return true;
     
317. }
     
318. 
     
319. void do_epoll(int epoll, epoll_event* wait_epevent, int wait_epevent_num
     
320.                           , int wait_time, char* buffer, int buf_size
     
321.                           , double& recv_size, double& send_size, unsigned int& out_sock_num)
     
322. {
     
323.         int wait_num = epoll_wait(epoll, wait_epevent, wait_epevent_num, wait_time);
     
324. 
     
325.         for (int i=0; i<wait_num; i++)
     
326.         {
     
327.                 epitem* sockitem = (epitem*)wait_epevent[i].data.ptr;
     
328.                 if ( 0 != (wait_epevent[i].events & ~(EPOLLIN|EPOLLOUT)) )
     
329.                 {                       
     
330.                         // 非输入输出事件，全部关闭
     
331.                         goto fail;
     
332.                 }
     
333. 
     
334.                 if ((wait_epevent[i].events & EPOLLIN)
     
335.                         && !do_input(sockitem->sock, buffer, buf_size, recv_size))
     
336.                 {
     
337.                         // 输入事件处理失败
     
338.                         goto fail;
     
339.                 }
     
340. 
     
341.                 {
     
342.                         // 这里sock是有效的，为什么端口号会改变？？
     
343.                         unsigned short local_port = get_sock_local_port(sockitem->sock);
     
344.                         assert((0 == local_port) || (local_port == sockitem->local_port));
     
345. 
     
346.                         unsigned short remote_port = get_sock_remote_port(sockitem->sock);
     
347.                         assert((0 == remote_port) || (remote_port == sockitem->remote_port));
     
348.                 }
     
349. 
     
350.                 if ((wait_epevent[i].events & EPOLLOUT)
     
351.                         && !do_output(sockitem->sock, buffer, buf_size, send_size))
     
352.                 {
     
353.                         // 输出事件处理失败
     
354.                         goto fail;
     
355.                 }
     
356.                
     
357.                 {
     
358.                         // 这里sock是有效的，为什么端口号会改变？？
     
359.                         unsigned short local_port = get_sock_local_port(sockitem->sock);
     
360.                         assert((0 == local_port) || (local_port == sockitem->local_port));
     
361. 
     
362.                         unsigned short remote_port = get_sock_remote_port(sockitem->sock);
     
363.                         assert((0 == remote_port) || (remote_port == sockitem->remote_port));
     
364.                 }
     
365. 
     
366.                 // 事件过多时，随机关闭一些连接，仅仅是测试需要
     
367.                 if (rand()%5000 < (wait_num-1000))
     
368.                 {
     
369.                         goto fail;
     
370.                 }
     
371. 
     
372.                 goto ok;
     
373. 
     
374. fail:
     
375.                 if (0 != epoll_ctl(epoll, EPOLL_CTL_DEL, sockitem->sock, 0))
     
376.                 {
     
377.                         // 为什么会出现bad file descriptor?
     
378.                         //pirntf("epoll_ctl failed.[%d:%s]\r\n", errno, strerror(errno));
     
379.                 }
     
380.                 close(sockitem->sock);
     
381.                 delete sockitem;
     
382.                 out_sock_num++;
     
383. ok:
     
384.                 ;
     
385.         }
     
386. }
     
387. // 服务器端epoll数据处理
     
388. void* thread_server(void* param)
     
389. {
     
390.         printf("begin thread server...\r\n");
     
391. 
     
392.         // 创建接收和发送数据的缓冲区，所有sock公用，数据随机
     
393.         char*                buffer        = new char[buffer_size];
     
394.         assert(0 != buffer);
     
395. 
     
396.         // epoll中等待事件
     
397.         epoll_event* wait_epevent = new epoll_event[wait_epevent_num];
     
398.         assert (wait_epevent != 0);
     
399. 
     
400.         while(woking)
     
401.         {
     
402.                 do_epoll(ep_server, wait_epevent, wait_epevent_num, 1, buffer, buffer_size
     
403.                         , server_recv_size, server_send_size, server_outepoll_sock);
     
404.         }
     
405. 
     
406.         printf("end thread server...\r\n");
     
407.         return 0;
     
408. }
     
409. // 客户端epoll数据处理
     
410. void* thread_client(void* param)
     
411. {
     
412.         printf("begin thread client...\r\n");
     
413. 
     
414.         // 创建接收和发送数据的缓冲区，所有sock公用，数据随机
     
415.         char*                buffer        = new char[buffer_size];
     
416.         assert(0 != buffer);
     
417. 
     
418.         // epoll中等待事件
     
419.         epoll_event* wait_epevent = new epoll_event[wait_epevent_num];
     
420.         assert (wait_epevent != 0);
     
421. 
     
422.         while(woking)
     
423.         {
     
424.                 do_epoll(ep_client, wait_epevent, wait_epevent_num, 1, buffer, buffer_size
     
425.                         , client_recv_size, client_send_size, client_outepoll_sock);
     
426.         }
     
427. 
     
428.         delete[] buffer;
     
429.         delete[] wait_epevent;
     
430. 
     
431.         printf("end thread client...\r\n");
     
432.         return 0;
     
433. }
     
434. 
     
435. 
     
436. int main()
     
437. {
     
438.         rlimit rlim;
     
439.         if (0 == getrlimit(RLIMIT_NOFILE, &rlim))
     
440.         {
     
441.                 rlim.rlim_cur = rlim.rlim_max;
     
442.                 setrlimit(RLIMIT_NOFILE , &rlim);
     
443.         }
     
444. 
     
445.         // 创建服务器端epoll
     
446.         ep_server =  epoll_create(4096);
     
447.         assert(ep_server != -1);
     
448. 
     
449.         // 创建客户端端epoll
     
450.         ep_client =  epoll_create(4096);
     
451.         assert (ep_client != -1);
     
452. 
     
453.         pthread_t thread_id_connect;
     
454.         pthread_t thread_id_accept;
     
455.         pthread_t thread_id_client;
     
456.         pthread_t thread_id_server;
     
457. 
     
458.         woking = true;
     
459. 
     
460.         // 服务器处理线程
     
461.         if (0 != pthread_create(&thread_id_server, 0, thread_server, 0))
     
462.         {
     
463.                 return 0;
     
464.         }
     
465. 
     
466.         // 客户端处理线程
     
467.         if (0 != pthread_create(&thread_id_client, 0, thread_client, 0))
     
468.         {
     
469.                 return 0;
     
470.         }
     
471. 
     
472.         // 连接监听线程
     
473.         if (0 != pthread_create(&thread_id_accept, 0, thread_accept, 0))
     
474.         {
     
475.                 return 0;
     
476.         }
     
477.         // 建立连接线程
     
478.         if (0 != pthread_create(&thread_id_connect, 0, thread_connect, 0))
     
479.         {
     
480.                 return 0;
     
481.         }
     
482. 
     
483.         printf("begin epoll test...\r\n");
     
484. 
     
485.         while (1)
     
486.         {
     
487.                 // 打印一些统计信息，连接数和收发数据速度
     
488.                 // 连接数为什么不对？
     
489.                 printf("s[%d:%f:%f] c[%d:%f:%f]\r\n"
     
490.                         ,(server_inepoll_sock-server_outepoll_sock)
     
491.                         , server_recv_size/1024/1024,  server_send_size/1024/1024
     
492.                         ,(client_inepoll_sock-client_outepoll_sock)
     
493.                         , client_recv_size/1024/1024, client_send_size/1024/1024);
     
494. 
     
495.                 server_recv_size = 0.0d;
     
496.                 server_send_size = 0.0d;
     
497.                 client_recv_size = 0.0d;
     
498.                 client_send_size = 0.0d;
     
499. 
     
500.                 sleep(1);
     
501.         }
     
502. 
     
503.         sleep(-1);
     
504. 
     
505.         woking = false;
     
506. 
     
507.         close(ep_server);
     
508.         close(ep_client);
     
509. 
     
510.         pthread_join(thread_id_connect, 0);
     
511.         pthread_join(thread_id_accept, 0);
     
512.         pthread_join(thread_id_client, 0);
     
513.         pthread_join(thread_id_server, 0);
     
514. 
     
515.         printf("end epoll test...\r\n");
     
516. 
     
517.         return 0;
     
518. }

复制代码

flead

对你这个线程设置下 打开的句柄个数限制
setrlimit(RLIMIT_NOFILE, & rl)