请教关于epoll＋线程池的问题

yaoyefengyun

学习了lvyilong316前辈的 《彻底学会使用epoll》系列，我现在想做成多线程，但又不是一个线程监听事件，其它几个线程负责事务处理那种。用一个连接测试，没测出问题，但是模拟1000个客户端时，服务端老有Bad file descriptor错误。之前服务端采用单个线程时，测试程序也没测出问题。大家帮我看看是代码写错了呢？还是说连思路都不正确。

下面是服务端代码:

1. #include <stdio.h>
   
2. #include <stdlib.h>
   
3. #include <string.h>
   
4. #include <unistd.h>
   
5. #include <sys/types.h>
   
6. #include <netinet/in.h>
   
7. #include <sys/epoll.h>
   
8. #include <fcntl.h>
   
9. #include <errno.h>
   
10. #include <pthread.h>
    
11. 
    
12. #define MAX_EVENTS 2000
    
13. #define SERV_PORT 9127
    
14. #define MAXLINE 1024
    
15. 
    
16. #define handle_error(msg) \
    
17.         do { perror(msg); exit(EXIT_FAILURE); } while (0)
    
18. 
    
19. int epollfd, listenfd;
    
20. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
    
21. 
    
22. void setnonblocking(int fd)
    
23. {
    
24.         int flags, ret;
    
25. 
    
26.         flags = fcntl(fd, F_GETFL, 0);
    
27.         if (flags == -1)
    
28.                 handle_error("fcntl");
    
29. 
    
30.         ret = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    
31.         if (ret == -1)
    
32.                 handle_error("fcntl");
    
33. }
    
34. 
    
35. int create_and_bind(void)
    
36. {
    
37.         int listenfd, err;
    
38.         struct sockaddr_in servaddr;
    
39. 
    
40.         listenfd = socket(AF_INET, SOCK_STREAM, 0);
    
41.         if (listenfd == -1)
    
42.                 handle_error("socket");
    
43. 
    
44.         bzero(&servaddr, sizeof(servaddr));
    
45.         servaddr.sin_family = AF_INET;
    
46.         servaddr.sin_port = htons(SERV_PORT);
    
47.         servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
48. 
    
49.         err = bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    
50.         if (err == -1)
    
51.                 handle_error("bind");
    
52. 
    
53.         return listenfd;
    
54. }
    
55. 
    
56. void *thr_func(void *arg)
    
57. {
    
58.         int i, nfds, connfd, ret;
    
59.         socklen_t clilen;
    
60.         ssize_t n, nread, nwrite, data_size;
    
61.         char buf[MAXLINE];
    
62.         struct sockaddr_in cliaddr;
    
63.         struct epoll_event ev, events[MAX_EVENTS];
    
64. 
    
65.         for ( ; ; ) {
    
66.                 nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
    
67.                 if (nfds == -1)
    
68.                         handle_error("epoll_wait");
    
69. 
    
70.                 for (i = 0; i < nfds; ++i) {
    
71.                         if (events[i].data.fd == listenfd) {
    
72.                                 for ( ; ; ) {
    
73.                                         clilen = sizeof(cliaddr);
    
74.                                         connfd = accept(events[i].data.fd,
    
75.                                                 (struct sockaddr *)&cliaddr,
    
76.                                                 &clilen);
    
77.                                         if (connfd == -1) {
    
78.                                                 if (errno == EAGAIN)
    
79.                                                         break;
    
80.                                                 else if (errno == ECONNABORTED||
    
81.                                                         errno == EPROTO ||
    
82.                                                         errno == EINTR)
    
83.                                                         continue;
    
84.                                                 else
    
85.                                                         handle_error("accept");
    
86.                                         }
    
87. 
    
88.                                         if (connfd >= MAX_EVENTS) {
    
89.                                                 ret = close(connfd);
    
90.                                                 if (ret == -1)
    
91.                                                         handle_error("close");
    
92.                                                 continue;
    
93.                                         }
    
94. 
    
95.                                         setnonblocking(connfd);
    
96. 
    
97.                                         ev.data.fd = connfd;
    
98.                                         ev.events = EPOLLIN | EPOLLET;
    
99.                                         ret = epoll_ctl(epollfd,
    
100.                                                 EPOLL_CTL_ADD, connfd, &ev);
     
101.                                         if (ret == -1)
     
102.                                                 handle_error("epoll_ctl");
     
103.                                 }
     
104.                                 continue;
     
105.                         }
     
106.                         if (events[i].events & EPOLLIN) {
     
107.                                 n = 0;
     
108.                                 while ((nread = read(events[i].data.fd, buf + n,
     
109.                                                         MAXLINE)) > 0)
     
110.                                         n += nread;
     
111.                                 if (nread == 0) {
     
112.                                         ret = close(events[i].data.fd);
     
113.                                         if (ret == -1)
     
114.                                                 handle_error("close");
     
115.                                         continue;
     
116.                                 } else if (nread == -1 && errno != EAGAIN) {
     
117.                                         perror("read");
     
118.                                         continue;
     
119.                                 }
     
120. 
     
121.                                 ev.data.fd = events[i].data.fd;
     
122.                                 ev.events = EPOLLOUT | EPOLLET;
     
123.                                 ret = epoll_ctl(epollfd, EPOLL_CTL_MOD,
     
124.                                         events[i].data.fd, &ev);
     
125.                                 if (ret == -1)
     
126.                                         handle_error("epoll_ctl");
     
127.                         }
     
128.                         if (events[i].events & EPOLLOUT) {
     
129.                                 snprintf(buf, sizeof(buf),
     
130.                                         "HTTP/1.1 200 OK/r/n Content-Length: "
     
131.                                         "%d/r/n/r/nHello World",
     
132.                                         11);
     
133.                                 data_size = strlen(buf);
     
134.                                 n = data_size;
     
135.                                 while (n > 0) {
     
136.                                         nwrite = write(events[i].data.fd,
     
137.                                                 buf + data_size - n , n);
     
138.                                         if (nwrite == -1 &&
     
139.                                                 errno != EAGAIN)
     
140.                                                 handle_error("write");
     
141.                                         n -= nwrite;
     
142.                                 }
     
143. 
     
144.                                 ev.data.fd = events[i].data.fd;
     
145.                                 ev.events = EPOLLIN | EPOLLET;
     
146.                                 ret = epoll_ctl(epollfd, EPOLL_CTL_MOD,
     
147.                                         events[i].data.fd, &ev);
     
148.                                 if (ret == -1)
     
149.                                         handle_error("epoll_ctl");
     
150.                         }
     
151.                 }
     
152.         }
     
153. }
     
154. 
     
155. int main(void)
     
156. {
     
157.         int i, ret;
     
158.         struct epoll_event ev;
     
159.         pthread_t tid[4];
     
160. 
     
161.         listenfd = create_and_bind();
     
162. 
     
163.         setnonblocking(listenfd);
     
164. 
     
165.         ret = listen(listenfd, MAX_EVENTS);
     
166.         if (ret == -1)
     
167.                 handle_error("listen");
     
168. 
     
169.         epollfd = epoll_create1(0);
     
170.         if (epollfd == -1)
     
171.                 handle_error("epoll_create");
     
172. 
     
173.         ev.events = EPOLLIN | EPOLLET;
     
174.         ev.data.fd = listenfd;
     
175.         ret = epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev);
     
176.         if (ret == -1)
     
177.                 handle_error("epoll_ctl");
     
178. 
     
179.         for (i = 0; i < 4; i++) {
     
180.                 ret = pthread_create(&tid[i], 0, thr_func, 0);
     
181.                 if (ret != 0)
     
182.                         handle_error("pthread_crete");
     
183.         }
     
184. 
     
185.         for (i = 0; i < 4; i++) {
     
186.                 ret = pthread_join(tid[i], 0);
     
187.                 if (ret != 0)
     
188.                         handle_error("pthread_jion");
     
189.         }
     
190. 
     
191.         exit(EXIT_SUCCESS);
     
192. }

复制代码

下面是客户端测试代码:

1. #include <stdio.h>
   
2. #include <stdlib.h>
   
3. #include <string.h>
   
4. #include <errno.h>
   
5. #include <unistd.h>
   
6. #include <sys/types.h>
   
7. #include <sys/socket.h>
   
8. #include <netinet/in.h>
   
9. #include <arpa/inet.h>
   
10. #include <pthread.h>
    
11. #include <signal.h>
    
12. 
    
13. #define MAXLINE 8192
    
14. #define LISTENQ 1024
    
15. #define SERV_PORT 9127
    
16. 
    
17. #define handle_error(msg) \
    
18.         do { perror(msg); exit(EXIT_FAILURE); } while (0)
    
19. 
    
20. #define err_quit(msg) \
    
21.         do { fprintf(stderr, msg); exit(EXIT_FAILURE); } while (0)
    
22. 
    
23. void str_cli(FILE *fp, int sockfd)
    
24. {
    
25.         pid_t pid;
    
26.         int n, ret;
    
27.         char sendline[MAXLINE + 1], recvline[MAXLINE + 1];
    
28. 
    
29.         pid = fork();
    
30.         switch (pid) {
    
31.         case -1:
    
32.                 handle_error("fork");
    
33.                 break;
    
34.         case 0:
    
35.                 while ((n = read(sockfd, recvline, MAXLINE)) > 0) {
    
36.                         recvline[n] = 0;
    
37.                         fprintf(stderr, recvline);
    
38.                 }
    
39. 
    
40.                 kill(getppid(), SIGTERM);
    
41.                 exit(0);
    
42.                 break;
    
43.         default:
    
44.                 while (fgets(sendline, MAXLINE, fp))
    
45.                         write(sockfd, sendline, strlen(sendline));
    
46.                 break;
    
47.         }
    
48. 
    
49.         ret = shutdown(sockfd, SHUT_WR);
    
50.         if (ret == -1)
    
51.                 handle_error("shutdown");
    
52. }
    
53. 
    
54. void *thr_func(void *arg)
    
55. {
    
56.         int sockfd, ret;
    
57.         ssize_t s;
    
58.         struct sockaddr_in servaddr;
    
59.         char buf[MAXLINE];
    
60. 
    
61.         sockfd = socket(AF_INET, SOCK_STREAM, 0);
    
62.         if (sockfd == -1)
    
63.                 handle_error("socket");
    
64. 
    
65.         bzero(&servaddr, sizeof(servaddr));
    
66.         servaddr.sin_family = AF_INET;
    
67.         servaddr.sin_port = htons(SERV_PORT);
    
68. 
    
69.         ret = inet_pton(AF_INET, "127.0.0.1", &servaddr.sin_addr);
    
70.         if (ret == -1)
    
71.                 handle_error("inet_pton");
    
72. 
    
73.         ret = connect(sockfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    
74.         if (ret == -1)
    
75.                 handle_error("connect");
    
76. 
    
77.         s = write(sockfd, "echo", 4);
    
78.         if (s == -1)
    
79.                 fprintf(stderr, "%ld, write %s\n",
    
80.                         (long)pthread_self(), strerror(errno));
    
81.         ret = shutdown(sockfd, SHUT_WR);
    
82.         if (ret == -1)
    
83.                 handle_error("shutdown");
    
84.         s = read(sockfd, buf, MAXLINE);
    
85.         if (s == -1)
    
86.                 fprintf(stderr, "read %s\n", strerror(errno));
    
87. 
    
88.         return NULL;
    
89. }
    
90. 
    
91. int main(void)
    
92. {
    
93.         int i, ret;
    
94.         pthread_t tid[1000];
    
95. 
    
96.         for (i = 0; i < 1000; i++) {
    
97.                 ret = pthread_create(&tid[i], 0, thr_func, 0);
    
98.                 if (ret != 0)
    
99.                         handle_error("pthread_create");
    
100.         }
     
101. 
     
102.         for (i = 0; i < 1000; i++) {
     
103.                 ret = pthread_join(tid[i], NULL);
     
104.                 if (ret != 0)
     
105.                         handle_error("pthread_join");
     
106.         }
     
107.         exit(EXIT_SUCCESS);
     
108. }
     
109. 
     

复制代码

bfdhczw

为何要在4个线程中对同一个fd做epoll_wait呢？

yaoyefengyun

因为如果只有一个线程，当线程执行epoll_wait以外的操作时，有可能会有新的I/O事件到来。四个线程的话，当一两个线程处理其它工作的时候，其它线程依然阻塞在epoll_wait,能够及时处理新到来的I/O事件。
可能我的想法不太对，处理比较复杂的业务时，或许应该一个线程接收I/O事件，接收到的事件放到一个队列，然后让线程池中的线程来处理这个队列的事件？回复 2# bfdhczw


   

bfdhczw

回复 3# yaoyefengyun

或许应该一个线程接收I/O事件，接收到的事件放到一个队列，然后让线程池中的线程来处理这个队列的事件？

我也倾向于这种做法。
   

yaoyefengyun

再请教一个问题，这种非阻塞的情况下，不完整write应该怎么处理呢，轮询直到write完吗，还是说重新epoll_mod OUT事件呢?感觉如果轮询，有点浪费cpu资源，重新epoll_mod OUT的话，似乎又和期望的高效率相违背，有点纠结。回复 4# bfdhczw


   

bfdhczw

回复 5# yaoyefengyun
其实我也是初学者，感觉这种情况等待下一次事件应该好些吧。

   

yaoyefengyun

嗯，谢谢！回复 6# bfdhczw