select+多线程 问题好纠结，愁死了

guojinshuai

代码是这样写的：当主线程发现有连接来时加入FD_SET，select到可读时，加一个读的任务到队列tasklist中，然后有3个子线程循环着从队列取任务read数据。
但是现在有个问题：只要客户端一发数据，服务端的tasklist.size()立马就上千，导致后续很多问题。
哪位热心朋友帮忙看看代码~~3Q、、

服务端：

1. 
   
2. #include  <unistd.h>
   
3. #include  <sys/types.h>       /* basic system data types */
   
4. #include  <sys/socket.h>      /* basic socket definitions */
   
5. #include  <netinet/in.h>      /* sockaddr_in{} and other Internet defns */
   
6. #include  <arpa/inet.h>       /* inet(3) functions */
   
7. #include  <stdlib.h>
   
8. #include  <errno.h>
   
9. #include  <stdio.h>
   
10. #include  <string.h>
    
11. #include  <fcntl.h>
    
12. #include  <sys/select.h>
    
13. #include  <pthread.h>
    
14. #include  <iostream>
    
15. #include  <list>
    
16. #include <signal.h>
    
17. using namespace std;
    
18. 
    
19. 
    
20. pthread_mutex_t lock;
    
21. 
    
22. enum falgrw{readflag = 0,writeflag};
    
23. 
    
24. typedef struct{
    
25. int fd;
    
26. int op;
    
27. }Task;
    
28. 
    
29. void *func( void *arg );
    
30. Task *create_task(int socket);
    
31. int clientSockFds[FD_SETSIZE]={0};
    
32. fd_set allset;
    
33. list<Task> tasklist(0);
    
34. int  main()
    
35. {
    
36.        
    
37.         int guo=0;
    
38.         pthread_t thread[3];
    
39.         for(guo=0;guo<3;guo++)
    
40.         {
    
41.                 pthread_create(&thread[guo],NULL,func,NULL);
    
42.         }
    
43. 
    
44.     int  listenfd = 0;
    
45.         int    connfd = 0;
    
46.     int serverPort= 8888;
    
47.     int   listenq = 5;
    
48.     socklen_t socklen = 0;
    
49.     int opt = 1;
    
50.     int nready = 0;       
    
51.     int maxfd  = 0;       
    
52.         int maxi = -1;
    
53. 
    
54.         char buf[50]= {0};
    
55.     struct sockaddr_in cliaddr, servaddr;
    
56.     socklen = sizeof(struct sockaddr_in);
    
57. 
    
58.     memset(&servaddr, 0, sizeof(servaddr));
    
59.     memset(&cliaddr,  0, sizeof(cliaddr));
    
60.        
    
61.     servaddr.sin_family = AF_INET;
    
62.     servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
63.     servaddr.sin_port = htons(serverPort);
    
64. 
    
65.     listenfd = socket(AF_INET, SOCK_STREAM, 0);
    
66.     if (listenfd < 0)
    
67.         {
    
68.         perror("socket error");
    
69.         return -1;
    
70.     }
    
71.         if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt)) < 0)
    
72.     {
    
73.         perror("setsockopt error");   
    
74.     }
    
75.     if (bind(listenfd, (struct sockaddr *) &servaddr, socklen) < 0)
    
76.         {
    
77.         perror("bind error");
    
78.         return -1;
    
79.     }
    
80.     if (listen(listenfd, listenq) < 0)
    
81.         {
    
82.         perror("listen error");   
    
83.         return -1;
    
84.     }
    
85.        
    
86.     FD_ZERO(&allset);
    
87.     FD_SET(listenfd, &allset);
    
88.     maxfd = listenfd;   
    
89. 
    
90.         for (int j = 0; j< FD_SETSIZE; j++)
    
91.         {
    
92.         clientSockFds[j] = -1;                
    
93.         }
    
94.        
    
95.     printf("echo server startup,listen on port:%d\n", serverPort);
    
96.         printf("max connection: %d\n", FD_SETSIZE);
    
97.         while(1)  
    
98.     {
    
99.         fd_set rset = allset;
    
100.         nready = select(maxfd + 1, &rset, NULL, NULL, NULL);
     
101.                 //printf("nready = %d\n",nready);
     
102.         if (nready < 0)
     
103.         {
     
104.             perror("select error");
     
105.             continue;
     
106.         }
     
107.         if (FD_ISSET(listenfd, &rset))
     
108.         {
     
109.             connfd = accept(listenfd, (struct sockaddr*) &cliaddr, &socklen);
     
110.             if (connfd < 0)
     
111.             {
     
112.                 perror("accept error");
     
113.                 continue;
     
114.             }
     
115.                        
     
116.             sprintf(buf, "accept form %s:%d: connfd=%d\n", inet_ntoa(cliaddr.sin_addr), cliaddr.sin_port,connfd);
     
117.             printf(buf, "");
     
118. 
     
119.                         int i=0;
     
120.             for (i = 0; i< FD_SETSIZE; i++)
     
121.             {
     
122.                 if (clientSockFds[i] == -1)
     
123.                 {
     
124.                     clientSockFds[i] = connfd;
     
125.                     break;
     
126.                 }
     
127.             }
     
128.             if (i == FD_SETSIZE)
     
129.             {
     
130.                 fprintf(stderr, "too many connection, more than %d\n", FD_SETSIZE);
     
131.                 close(connfd);
     
132.                 break;
     
133.             }
     
134. 
     
135.             if (connfd > maxfd)
     
136.                 maxfd = connfd;
     
137.                                
     
138.             FD_SET(connfd, &allset);
     
139.         }
     
140.                 printf("main111()  tasklist.size():[%d]--->Thread:%d\n\n",tasklist.size(),pthread_self());       
     
141.                 for(int m=0;m<1024 && clientSockFds[m] != -1 ;m++)
     
142.                 {
     
143.                         if (FD_ISSET(clientSockFds[m], &rset))
     
144.                         {
     
145.                                 pthread_mutex_lock(&lock);
     
146.                                 Task *task = create_task(clientSockFds[m]);
     
147.                                 tasklist.push_back(*task);
     
148.                                
     
149.                                 printf("main222()  task->fd:%d--->tasklist.size():[%d]--->Thread:%d\n",task->fd,tasklist.size(),pthread_self());       
     
150.                                 pthread_mutex_unlock(&lock);
     
151.                         }
     
152.                
     
153.                 }
     
154.     }
     
155.         close(listenfd);
     
156.         return 0;
     
157. }
     
158. 
     
159. Task *create_task(int socket)
     
160. {
     
161.          Task *task = (Task *)malloc(sizeof(Task));
     
162.          task->fd=socket;
     
163.          task->op= readflag;
     
164.         return task;
     
165. }
     
166. 
     
167. void *func( void *arg )
     
168. {
     
169.         char buffer[1024]={0};
     
170.         Task task;
     
171.         pthread_detach(pthread_self());
     
172.         while(1)
     
173.         {
     
174.                 pthread_mutex_lock(&lock);
     
175.                 if(tasklist.size()>0)
     
176.                 {
     
177.                        
     
178.                         task = tasklist.front();
     
179.                         tasklist.pop_front();
     
180.                         printf("task.fd:%d----->tasklist.size():[%d]--->Thread:%d\n",task.fd,tasklist.size(),pthread_self());
     
181.                         exit(0);
     
182.                         int xxx=read(task.fd,buffer,sizeof(buffer));
     
183.                         if(xxx==0)
     
184.                         {
     
185.                                 printf("client %d close!\n",task.fd);
     
186.                                 FD_CLR(task.fd,&allset);
     
187.                                 close(task.fd);
     
188.                                 for (int i = 0; i< FD_SETSIZE; i++)
     
189.                                 {
     
190.                                         if (clientSockFds[i] == task.fd)
     
191.                                         {
     
192.                                                 clientSockFds[i] = -1;
     
193.                                                 break;
     
194.                                         }
     
195.                                 }
     
196.                                
     
197.                         }
     
198.                         else if(xxx>0)
     
199.                         {
     
200.                                 printf("fd:%d-->Thread:%d---->receive data:%s\n\n",task.fd,pthread_self(),buffer);
     
201.                         }
     
202.                         else
     
203.                         {
     
204.                                 printf("wrong!!\n");
     
205.                                 FD_CLR(task.fd,&allset);
     
206.                                 close(task.fd);
     
207.                                 for (int i = 0; i< FD_SETSIZE; i++)
     
208.                                 {
     
209.                                         if (clientSockFds[i] == task.fd)
     
210.                                         {
     
211.                                                 clientSockFds[i] = -1;
     
212.                                                 break;
     
213.                                         }
     
214.                                 }
     
215.                                
     
216.                         }
     
217.                        
     
218.                 }
     
219.                 pthread_mutex_unlock(&lock);
     
220.                 sleep(1);
     
221.         }
     
222. 
     
223.         return 0;
     
224. }
     
225. 
     

复制代码

客户端：

1. 
   
2. #include  <unistd.h>
   
3. #include  <sys/types.h>       /* basic system data types */
   
4. #include  <sys/socket.h>      /* basic socket definitions */
   
5. #include  <netinet/in.h>      /* sockaddr_in{} and other Internet defns */
   
6. #include  <arpa/inet.h>       /* inet(3) functions */
   
7. #include  <stdlib.h>
   
8. #include  <errno.h>
   
9. #include  <stdio.h>
   
10. #include  <string.h>
    
11. #include  <fcntl.h>
    
12. #include  <sys/select.h>
    
13. #include  <pthread.h>
    
14. #include  <iostream>
    
15. #include  <list>
    
16. using namespace std;
    
17. 
    
18. int main()
    
19. {
    
20.         int cfd; /* 文件描述符 */
    
21.         int recbytes;
    
22.         int sin_size;
    
23.         char buffer[1024]={0};   
    
24.         struct sockaddr_in s_add,c_add;
    
25.         unsigned short portnum=8888;  
    
26. 
    
27.         printf("Hello,welcome to client !\r\n");
    
28. 
    
29.         cfd = socket(AF_INET, SOCK_STREAM, 0);
    
30.         if(-1 == cfd)
    
31.         {
    
32.                 printf("socket fail ! \r\n");
    
33.                 return -1;
    
34.         }
    
35.         printf("socket ok !\r\n");
    
36.        
    
37.         bzero(&s_add,sizeof(struct sockaddr_in));
    
38.         s_add.sin_family=AF_INET;
    
39.         s_add.sin_addr.s_addr= inet_addr("127.0.0.1");
    
40.         s_add.sin_port=htons(portnum);
    
41. 
    
42.         if(-1 == connect(cfd,(struct sockaddr *)(&s_add), sizeof(struct sockaddr)))
    
43.         {
    
44.                 printf("connect fail !\r\n");
    
45.                 return -1;
    
46.         }
    
47.         printf("connect ok !\r\n");
    
48. 
    
49.         int j=0;
    
50.         struct timeval tv;  
    
51.         while(1)
    
52.         {
    
53.                
    
54.                 memset(buffer,0,sizeof(buffer));
    
55.                 sprintf(buffer,"%d%d",j,j);
    
56.                 if(-1 == (recbytes = write(cfd,buffer,strlen(buffer)+1)))
    
57.                 {
    
58.                         printf("write data fail !\r\n");
    
59.                         return -1;
    
60.                 }
    
61.                 printf("write ok.buffer=[%s]\n",buffer);
    
62.                 sleep(1);
    
63.                 j++;
    
64.                
    
65.                 //test select   
    
66.          tv.tv_sec = 0;  
    
67.          tv.tv_usec = 1000;  
    
68.                  int ret = select(0, NULL, NULL, NULL, &tv);  
    
69.          if (-1 == ret)  
    
70.          {  
    
71.             printf("wweeeeeee\n");
    
72.          }
    
73.        
    
74.         }
    
75.         close(cfd);
    
76.         return 0;
    
77. }
    

复制代码

weishuo1999

你的代码逻辑，应该是来了一个数据包，就建一个task，而不是一个连接建一个task，所以客户端循环发送的时候，自然服务端的tasklist瞬间就上升到1024了

guojinshuai

回复 2# weishuo1999


    懂了，一般来说，你们是一个连接一个task，还是一个数据包一个task? 哪种好点呢？

guojinshuai

为啥我用epoll，用相同得思路写代码，就没select那个问题呢？？？

1. 
   
2. #include  <unistd.h>
   
3. #include  <sys/types.h>       /* basic system data types */
   
4. #include  <sys/socket.h>      /* basic socket definitions */
   
5. #include  <netinet/in.h>      /* sockaddr_in{} and other Internet defns */
   
6. #include  <arpa/inet.h>       /* inet(3) functions */
   
7. #include  <stdlib.h>
   
8. #include  <errno.h>
   
9. #include  <stdio.h>
   
10. #include  <string.h>
    
11. #include  <fcntl.h>
    
12. #include  <sys/select.h>
    
13. #include  <pthread.h>
    
14. #include  <iostream>
    
15. #include <sys/epoll.h>
    
16. #include  <list>
    
17. #include <signal.h>
    
18. using namespace std;
    
19. 
    
20. 
    
21. pthread_mutex_t lock;
    
22. 
    
23. enum falgrw{readflag = 0,writeflag};
    
24. 
    
25. typedef struct{
    
26.         int fd;
    
27.         int op;
    
28. }Task;
    
29. 
    
30. void *func( void *arg );
    
31. Task *create_task(int socket);
    
32. struct epoll_event ev,events[20];
    
33. int epfd=0;
    
34. 
    
35. list<Task> tasklist(0);
    
36. int  main()
    
37. {
    
38.        
    
39.         int guo=0;
    
40.         pthread_t thread[3];
    
41.         for(guo=0;guo<3;guo++)
    
42.         {
    
43.                 pthread_create(&thread[guo],NULL,func,NULL);
    
44.         }
    
45. 
    
46.         int  listenfd = 0;
    
47.         int    connfd = 0;
    
48.         int serverPort= 8888;
    
49.         int   listenq = 5;
    
50.         socklen_t socklen = 0;
    
51.         int opt = 1;
    
52.         int nready = 0;       
    
53.         int maxfd  = 0;       
    
54.         int maxi = -1;
    
55. 
    
56.         char buf[50]= {0};
    
57.         struct sockaddr_in cliaddr, servaddr;
    
58.         socklen = sizeof(struct sockaddr_in);
    
59. 
    
60.         memset(&servaddr, 0, sizeof(servaddr));
    
61.         memset(&cliaddr,  0, sizeof(cliaddr));
    
62.        
    
63.         servaddr.sin_family = AF_INET;
    
64.         servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
65.         servaddr.sin_port = htons(serverPort);
    
66. 
    
67.         listenfd = socket(AF_INET, SOCK_STREAM, 0);
    
68.         if (listenfd < 0)
    
69.         {
    
70.                 perror("socket error");
    
71.                 return -1;
    
72.         }
    
73.         if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt)) < 0)
    
74.         {
    
75.                 perror("setsockopt error");   
    
76.         }
    
77.         if (bind(listenfd, (struct sockaddr *) &servaddr, socklen) < 0)
    
78.         {
    
79.                 perror("bind error");
    
80.                 return -1;
    
81.         }
    
82.         if (listen(listenfd, listenq) < 0)
    
83.         {
    
84.                 perror("listen error");   
    
85.                 return -1;
    
86.         }
    
87.        
    
88.         //epoll LT
    
89.         epfd=epoll_create(256);
    
90.         ev.data.fd=listenfd;
    
91.         //设置要处理的事件类型
    
92.         ev.events=EPOLLIN;
    
93.         //注册epoll事件
    
94.         epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
    
95.        
    
96.         int nfds=0;
    
97.        
    
98.        
    
99.         printf("echo server startup,listen on port:%d\n", serverPort);
    
100.         while(1)  
     
101.         {
     
102.                
     
103.                 struct timeval tv;
     
104.                 tv.tv_sec = 1;
     
105.                 tv.tv_usec = 0;
     
106. 
     
107.                 nfds=epoll_wait(epfd,events,20,500);
     
108.                 for(int i=0;i<nfds;++i)
     
109.                 {
     
110.                         if(events[i].data.fd==listenfd)
     
111.                         {                 
     
112.                                 connfd = accept(listenfd,(sockaddr *)&cliaddr, &socklen);
     
113.                                 if(connfd<0){
     
114.                                         perror("connfd<0");
     
115.                                         exit(1);
     
116.                                 }
     
117.                                 //设置用于读操作的文件描述符
     
118.                                 ev.data.fd=connfd;
     
119.                                 //设置用于注测的读操作事件
     
120.                                 ev.events=EPOLLIN|EPOLLET;
     
121.                                 //注册ev
     
122.                                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
     
123.                         }
     
124.                         else if(events[i].events&EPOLLIN)
     
125.                         {
     
126.                                 printf("You Can Read Data!\n");
     
127.                                 int sockfd=0;
     
128.                                 if ( (sockfd = events[i].data.fd) < 0)
     
129.                                 continue;
     
130.                                 pthread_mutex_lock(&lock);
     
131.                                 Task *task = create_task(sockfd);
     
132.                                 tasklist.push_front(*task);
     
133.                                 printf("ADD task;task.fd:%d----->tasklist.size():[%d]--->main()\n",task->fd,tasklist.size());
     
134.                                 pthread_mutex_unlock(&lock);
     
135.                         }
     
136.                         else if(events[i].events&EPOLLOUT)
     
137.                         {
     
138.                         }                        
     
139.                 }
     
140.         }
     
141.         close(listenfd);
     
142.         return 0;
     
143. }
     
144. 
     
145. Task *create_task(int socket)
     
146. {
     
147.         Task *task = (Task *)malloc(sizeof(Task));
     
148.         task->fd=socket;
     
149.         task->op= readflag;
     
150.         return task;
     
151. }
     
152. 
     
153. void *func( void *arg )
     
154. {
     
155.         char buffer[1024]={0};
     
156.         Task task;
     
157.         pthread_detach(pthread_self());
     
158.         while(1)
     
159.         {
     
160.                 pthread_mutex_lock(&lock);
     
161.                 if(tasklist.size()>0)
     
162.                 {
     
163.                         memset(buffer,0,1024);
     
164.                         task = tasklist.front();
     
165.                         //printf("task.fd:%d----->tasklist.size():[%d]--->Thread:%d\n",task.fd,tasklist.size(),pthread_self());
     
166.                         tasklist.pop_front();
     
167.                         int xxx=read(task.fd,buffer,sizeof(buffer));
     
168.                         if(xxx==0)
     
169.                         {
     
170.                                 printf("client %d close!\n",task.fd);
     
171.                                 printf("DEL task;task.fd:%d----->tasklist.size():[%d]--->Thread:%d---->receive data:%s\n\n",task.fd,tasklist.size(),pthread_self(),buffer);
     
172.                                 close(task.fd);
     
173.                                 epoll_ctl(epfd,EPOLL_CTL_DEL,task.fd,&ev);
     
174.                         }
     
175.                         else if(xxx>0)
     
176.                         {
     
177.                                 printf("DEL task;task.fd:%d----->tasklist.size():[%d]--->Thread:%d---->receive data:%s\n\n",task.fd,tasklist.size(),pthread_self(),buffer);
     
178.                         }
     
179.                         else
     
180.                         {
     
181.                                 printf("wrong!!\n");
     
182.                                 close(task.fd);
     
183.                                 epoll_ctl(epfd,EPOLL_CTL_DEL,task.fd,&ev);
     
184.                         }
     
185.                        
     
186.                 }
     
187.                 pthread_mutex_unlock(&lock);
     
188.                 sleep(1);
     
189.         }
     
190. 
     
191.         return 0;
     
192. }
     
193. 
     

复制代码

为啥我用epoll，用相同得思路写代码，就没select那个问题呢？？？

asdf2110

是不是因为在 accept 后的 fd ，使用了ET触发？
改成 LT 试试呢？

1.                                 //设置用于读操作的文件描述符
   
2.                                 ev.data.fd=connfd;
   
3.                                 //设置用于注测的读操作事件
   
4.                                 ev.events=EPOLLIN|EPOLLET;

复制代码

guojinshuai

[code]                                //设置用于读操作的文件描述符

                                ev.data.fd=connfd;

                                //设置用于注测的读操作事件

                                ev.events=EPOLLIN;[/code


改成这样，也没有select那个问题！！
   

guojinshuai

回复 5# asdf2110


    我本来用的就是epoll LT模式。。。

guojinshuai

回复 5# asdf2110


    看错了，我改改试试。

   epoll改成LT之后，现象跟select一样了！！！！求解释，求解决办法！！！！！

hellioncu

一个socket可读，在没有把数据read完之前，一直可读(水平触发)，所以你得在放到tasklist之后，需要把该socket从读fd_set中去掉，read完再加入fd_set。

BTW：采用你这样线程分工的很少见。

guojinshuai

回复 9# hellioncu


    恩，原因是这个，按你说的方法解决了下，确实没问题了，非常感谢啊。

   那更好的利用多线程你们一般怎么做的呀，我是初学者。我这样利用多线程效率不高吧？