一次select一个socket的测试结果

硬件：刀片2

接收端代码： 1024个socketl建立连接后，创建8个线程接收数据，每个接收线程处理128个socket, 每个接收线程对每个socket先select, 再recv（2048字节）300次。

发送端代码：4个发送端，每个发送端256个线程发送数据，一共1024个socket.

加了select后， 性能和不加select直接recv比起来变化很小。但是加了select后，接收端和发送端程序不再因为阻塞时间超时而出错。程序运行的稳定性增加了。

脚本 bw.sh 统计接收端的带宽为15.78 Gb/s。

接收端的接收队列长度为2MB左右。接收端接收速度还是慢， 不过相比每个select多个socket的工作模式，接收速度已经提高了很多。

发送端代码： server1bak.c ，和上一篇博客一样

统计带宽代码：bw.sh 和上一篇一样

接收端代码： client1_multi_select.c

 1 #include <stdio.h>
  2 #include <stdlib.h>
  3 
  4 #include<pthread.h>
  5 #include <unistd.h>
  6 #include <time.h>
  7 
  8 #include<sys/ioctl.h>
  9 #include <sys/socket.h>
 10 #include <arpa/inet.h>
 11 #include <netinet/in.h>
 12 #include<sys/time.h>
 13 #include<string>
 14 
 15 #define PORT 33333
 16 
 17 #define SOCKNUM 1024
 18 #define THREAD_NUM 8
 19 #define SOCKET_PER_THREAD 128
 20 #define SERVER_NUM 4
 21 #define MSGSIZE 2048
 22 
 23 typedef struct{
 24     int sock[SOCKET_PER_THREAD];
 25 }ARG;
 26 
 27 int SetSocketOptions(int fd)
 28 {
 29     int sockopt = 0;
 30     int SOCKET_ERROR = -1;
31     static const int c_so_rcvbuf = 256*1024;
 32     static const int c_so_sndbuf = 256*1024;
 33 
 34     sockopt = 1;
 35     if ( setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&sockopt, sizeof(sockopt)) == SOCKET_ERROR )
 36     {
 37         perror("set reuseaddr error");
 38         return -1;
 39     }
 40 
 41     sockopt = c_so_sndbuf;
 42     if ( setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sockopt, sizeof(sockopt)) == SOCKET_ERROR )
 43     {
 44         perror("set so_sndbuf error");
 45         return -1;
 46     }
 47 
 48     sockopt = c_so_rcvbuf;
 49     if ( setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&sockopt, sizeof(sockopt)) == SOCKET_ERROR )
 50     {
 51         perror("set so_rcvbuf error");
 52         return -1;
 53     }
 54 }
 55 
 56 
 57 int poll(int socket)
 58 {
 59     fd_set recv_fds;
 60     struct timeval c_select_timeout = {0, 0};
61     FD_ZERO(&recv_fds);
 62     FD_SET(socket, &recv_fds);
 63 
 64     int sel = select(socket+1, &recv_fds, NULL, NULL, &c_select_timeout);
 65     if(sel == 0)
 66     {
 67         return 0;
 68     }
 69     else if(sel > 0)
 70     {
 71         if( FD_ISSET(socket, &recv_fds) ) // socket is readable
 72         {
 73             int length;
 74             int status = ioctl(socket, FIONREAD, &length);
 75             if(status == -1)
 76             {
 77                 printf("Error reading input size
");
 78             }
 79             if(length)
 80             {
 81                 //printf("data length in socket buffer: %lf MB
", length/1024.0/1024.0);
 82                 return 1;
 83             }
 84             else
 85             {
 86                 printf("Nothing to read, eof??
");
 87                 if(socket != -1)
 88                 {
 89                     close(socket);
 90                     socket = -1;
 91                 }
 92                 perror("socket flagged but no data available probable EOF");
 93                 return 0;
 94             }
 95 
 96         }
 97         else
 98         {
 99             if(socket != -1)
100             {
101                 close(socket);
102                 socket = -1;
103             }
104             perror("FD_ISSET == zero");
105         }
106 
107     }
108     else
109     {
110         if(socket != -1)
111         {
112             close(socket);
113             socket = -1;
114         }
115         perror("select<0 error");
116     }
117 }
118 
119 
120 
121 int recvdata(int sock, char *buffer)
122 {
123     int msgsize = MSGSIZE;
124     int ret;
125     int nrecv=0;
126     while (nrecv < msgsize)
127     {
128         ret = recv(sock, buffer, msgsize-nrecv, 0);
129         if (ret < 0)
130         {
131             perror("recv fail");
132             exit(1);
133         }
134         else
135         {
136             nrecv += ret;
137         }
138     }
139     return nrecv;
140 }
141 
142 void *recvData(void *arg)
143 {
144     ARG* a = (ARG*)arg;
145     int *socket = a->sock;
146     char buffer[MSGSIZE] = "0";
147     int count = 0;
148     struct  timeval  start;
149     struct  timeval  end;
150     unsigned long timer;
151     gettimeofday(&start,NULL);
152 
153     while(1)
154     {
155         for(int i=0; i<SOCKET_PER_THREAD; i++)
156         {
157             if( poll(socket[i]) )
158             {
159             for(int num=0; num<300; num++)
160             recvdata(socket[i], buffer);
161             }
162 #if 0
163             count++;
164             gettimeofday(&end,NULL);
165             timer = 1000000 * (end.tv_sec-start.tv_sec)+ end.tv_usec-start.tv_usec;
166             if(timer % 5000000==0)
167             {
168                 printf("timer = %ld us, %lf Gb/s
",timer, count*2048.0/timer/1024*8);
169             }
170 #endif
171         }
172     }
173     return 0;
174 }
175 
176 
177 int main()
178 {
179     int sock[SERVER_NUM][SOCKNUM/SERVER_NUM];
180     struct sockaddr_in addr_ser[SERVER_NUM];
181     struct sockaddr_in addr_cli[SERVER_NUM][SOCKNUM/SERVER_NUM];
182 
183     std::string local_ip("192.168.250.141");
184 
185     std::string server_ip[SERVER_NUM] = {"192.168.250.146", "192.168.250.147", "192.168.250.142", "192.168.250.143"};
186     //std::string server_ip[SERVER_NUM] = {"192.168.251.166", "192.168.251.167", "192.168.251.162", "192.168.251.163"}; 
187 //  std::string server_ip[SERVER_NUM] = {"192.168.251.163"}; 
188     for(int ser=0; ser < SERVER_NUM; ser++)
189     {
190     for(int i=0; i<SOCKNUM/SERVER_NUM; i++)
191     {
192         sock[ser][i] = socket(AF_INET, SOCK_STREAM, 0);
193         if(sock[ser][i] < 0)
194         {
195             printf("%d ", i);
196             perror("create socket fail");
197         }
198 
199         addr_ser[ser].sin_family = AF_INET;
200         addr_ser[ser].sin_port = htons(PORT);
201         addr_ser[ser].sin_addr.s_addr = inet_addr(server_ip[ser].c_str());
202 
203         addr_cli[ser][i].sin_family = AF_INET;
204         addr_cli[ser][i].sin_port = 0;
205         addr_cli[ser][i].sin_addr.s_addr = inet_addr(local_ip.c_str());
206 
207 
208         int sockopt = 1;
209         if ( setsockopt(sock[ser][i], SOL_SOCKET, SO_REUSEADDR, (char*)&sockopt, sizeof(sockopt)) == -1 )
210         {
211             perror("set reuseaddr error");
212             exit(1);
213         }
214 
215 
216 #if 0 
217 
218         if ( SetSocketOptions(sock[ser][i]) == -1)
219         {
220             perror("set socket options error");
221             exit(1);
222         }
223 #endif
224 
225         if( bind(sock[ser][i], (struct sockaddr*)&addr_cli[ser][i], sizeof(addr_cli[ser][i]) ) < 0 )
226         {
227             perror("TCP bind: ");
228             exit(1);
229         }
230         printf("bind ok!
");
231 
232         if(connect(sock[ser][i], (struct sockaddr*)&addr_ser[ser], sizeof(struct sockaddr)) < 0)
233         {
234             perror("connect fail:");
235             exit(1);
236         }
237         printf("connect ok!
");
238 
239     }
240 
241     }
242 
243 
244     int socket[SOCKNUM] ;
245     int count=0;
246     for(int i=0; i< SERVER_NUM; i++)
247     {
248         for(int j=0; j<SOCKNUM/SERVER_NUM; j++)
249         {
250             socket[count++] = sock[i][j];
251         }
252     }
253 
254     pthread_t tid[THREAD_NUM];
255     ARG a[THREAD_NUM];
256     for(int i=0; i<THREAD_NUM; i++)
257     {
258         for(int j=0; j<SOCKET_PER_THREAD; j++)
259         {
260         a[i].sock[j] = socket[i*SOCKET_PER_THREAD+j];
261         }
262         pthread_create(&tid[i], 0, recvData, (void *)&a[i]);
263     }
264 
265     for(int i=0; i<SOCKNUM; i++)
266     {
267         pthread_join(tid[i], 0);
268     }
269 
270     return 0;
271 }