一次select多个socket的测试结果

Posted 2020-09-15 小荷才楼尖尖角

接收端程序： 1024个socket建立连接后，创建8个线程收数据，每个线程处理128个socket, 每个线程先select这128个socket，然后用FD_ISSET对这128个socket进行检查和读出。

用脚本bw.sh 统计得到的接收端的总带宽为 0.114 Gb/s, 用 iftop 查看得到的接收端带宽在前几秒为3Gb/s, 然后下降到 112 Mb/s。

接收队列的长度为3.7MB。由此可见接收端很慢。

这种一次select多个socket的工作方式效率很低， 不应该采用。

发送端程序：server1bak.c 布置4个发送端

 1 #include <stdio.h>
  2 #include <stdlib.h>
  3 
  4 #include<pthread.h>
  5 #include <unistd.h>
  6 
  7 #include <sys/socket.h>
  8 #include <arpa/inet.h>
  9 #include <netinet/in.h>
 10 #include <netinet/tcp.h>
 11 #include<sys/time.h>
 12 #include<errno.h>
 13 #include<string.h>
 14 
 15 #define IP "192.168.250.147"
 16 #define PORT  33333
 17 #define SOCKNUM  256 
 18 #define PACKETSIZE 2048
 19 
 20 typedef struct {
 21     int sock;
 22 }ARG;
 23 
 24 pthread_t tid[SOCKNUM];
 25 
 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 
 58 
 59 int senddata(int sock)
 60 {
61     int ret;
 62     char temp[PACKETSIZE+1] = "I want to know!";
 63     int size_left=2048;
 64     while( size_left > 0)
 65     {
 66         ret = send(sock, temp, size_left, 0);
 67         if (ret < 0)
 68         {
 69             perror("send fail");
 70             exit(1);
 71         }
 72         size_left -= ret;
 73     }
 74 
 75     return size_left;
 76 }
 77 
 78 
 79 int read_cwnd(int tcp_work_socket)
 80 {
 81     struct tcp_info info;
 82     socklen_t  tcp_info_length = sizeof(tcp_info);
 83     if ( getsockopt(tcp_work_socket, SOL_TCP, TCP_INFO, (void *)&info, &tcp_info_length) == 0 ) {
 84         printf(" cwnd:%u, snd_ssthresh:%u, rtt:%u, rtt_d:%u\n",
 85                 info.tcpi_snd_cwnd,
 86                 info.tcpi_snd_ssthresh,
 87                 info.tcpi_rtt,
 88                 info.tcpi_rttvar
 89            );
90     }
 91     return 0;
 92 }
 93 
 94 
 95 void *sendData(void *arg)
 96 {
 97 #if 1 
 98     ARG *a = (ARG *)arg;
 99     int accept_sock = a->sock;
100 
101     long count = 0;
102     struct  timeval  start;
103     struct  timeval  end;
104     unsigned long timer=0;
105     gettimeofday(&start,NULL);
106 
107 
108     while(1){
109         senddata(accept_sock);
110         usleep(900);
111         if(pthread_self()== tid[0])
112         {
113 //      FILE* fpointer = fopen("result.out", "a+");
114         count++;
115         gettimeofday(&end,NULL);
116         timer = 1000000 * (end.tv_sec-start.tv_sec)+ end.tv_usec-start.tv_usec;
117         if(timer%2000== 0)
118         {
119             printf("count: %ld, socket: %ld,  %lf s, %lf Gb/s, ", count, accept_sock,  timer/1000000.0, count*2048.0/timer/1024*8);
120             read_cwnd(accept_sock);
121         }
122 
123 #if 1 
124         if(timer > 3600000000)
125         {
126             printf("before close: timer: %ld\n", timer);
127             close(accept_sock);
128 //          fclose(fpointer);
129             break;
130         }
131 #endif
132         }
133     }
134     return 0;
135 #endif
136 
137 }
138 
139 
140 
141 int main()
142 {
143     int accept_sock[SOCKNUM];
144     struct sockaddr_in addr_ser;
145 
146 
147     int sock = socket(AF_INET, SOCK_STREAM, 0);
148     if (sock < 0)
149     {
150         perror("create sock fail");
151         exit(1);
152     }
153 
154     addr_ser.sin_family = AF_INET;
155     addr_ser.sin_port = htons(PORT);
156 //  addr_ser.sin_addr.s_addr = inet_addr(IP);
157     addr_ser.sin_addr.s_addr = htonl(INADDR_ANY);
158 
159 
160     int sockopt = 1;
161     if (setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, (char *)&sockopt, sizeof(int)) < 0)
162     {
163         perror("setsockopt fail");
164         exit(1);
165     }
166 
167     if (bind(sock, (struct sockaddr*)&addr_ser, sizeof(struct sockaddr)) < 0)
168     {
169         perror("bind fail");
170         exit(1);
171     }
172 
173 
174 #if 0 
175     if ( SetSocketOptions(sock) == -1)
176     {
177         perror("set socket options error");
178         exit(1);
179     }
180 #endif
181 
182     if (listen(sock, 2000) < 0)
183     {
184         perror("listen fail");
185         exit(1);
186     }
187 
188 
189    for(int i=0; i<SOCKNUM; i++)
190    {
191 
192     accept_sock[i] = accept(sock, 0, 0);
193     if (accept_sock[i] < 0)
194     {
195         perror("accept fail");
196         exit(1);
197     }
198     printf("accept ok!\n");
199    }
200 
201 
202 #if 1 
203 
204    //static extern  pthread_t tid[SOCKNUM];
205    ARG a[SOCKNUM];
206    for(int i=0; i<SOCKNUM; i++){
207    a[i].sock = accept_sock[i];
208    pthread_create(&tid[i], 0, sendData, (void *)&a[i]);
209    }
210 #endif
211 
212 #if 1 
213    for(int i=0; i<SOCKNUM; i++)
214    {
215        pthread_join(tid[i], 0);
216    }
217 #endif
218 
219    return 0;
220 }

 

接收端代码：client1_multi_select2.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 int recvdata(int sock, char *buffer)
 57 {
 58     int msgsize = MSGSIZE;
 59     int ret;
 60     int nrecv=0;
61     while (nrecv < msgsize)
 62     {
 63         ret = recv(sock, buffer, msgsize-nrecv, 0);
 64         if (ret < 0)
 65         {
 66             perror("recv fail");
 67             exit(1);
 68         }
 69         else
 70         {
 71             nrecv += ret;
 72         }
 73     }
 74     return nrecv;
 75 }
 76 
 77 int max(int a, int b)
 78 {
 79     if(a >= b)
 80         return a;
 81     else
 82         return b;
 83 }
 84 
 85 
 86 
 87 
 88 int  poll_and_recv(fd_set recv_fds, int maxfd, int* socket)
89 {
 90 #if 1 
 91     char buffer[MSGSIZE] = "0";
 92     struct timeval c_select_timeout = {0, 0};
 93     int sel = select(maxfd+1, &recv_fds, NULL, NULL, &c_select_timeout);
 94     if(sel == 0)
 95     {
 96         return 0;
 97     }
 98     else if(sel > 0)
 99     {
100         for(int i=0; i<SOCKET_PER_THREAD; i++)
101         {
102             if( FD_ISSET(socket[i], &recv_fds) ) // socket is readable
103             {
104                 int length;
105                 int status = ioctl(socket[i], FIONREAD, &length);
106                 if(status == -1)
107                 {
108                     printf("Error reading input size\n");
109                 }
110                 if(length)
111                 {
112                     //printf("data in socket %d : %d\n", socket[i], length);
113 //                  for(int num=0; num<300; num++)
114 //                  {
115                         recvdata(socket[i], buffer);
116 //                  }
117                 }
118                 else // length==0
119                 {
120                     printf("Nothing to read, eof??\n");
121                     if(socket[i] != -1)
122                     {
123                         close(socket[i]);
124                         socket[i] = -1;
125                     }
126                     perror("socket flagged but no data available probable EOF");
127                 }
128 
129             } // FD_ISSET != 0
130         } // i < SOCKET_PER_THREAD 
131     } // sel > 0
132     else // sel < 0
133     {
134         for (int i=0; i<SOCKET_PER_THREAD; i++)
135         {
136             if(socket[i] != -1)
137             {
138                 close(socket[i]);
139                 socket[i] = -1;
140             }
141         }
142         perror("select<0 error");
143         return 0;
144     }
145 146 #endif
147     return 1;
148 
149 }
150 
151 
152 
153 void *recvData(void *arg)
154 {
155     ARG* a = (ARG*)arg;
156     int *socket = a->sock;
157     char buffer[MSGSIZE] = "0";
158     int count = 0;
159 
160     int maxfd = 0;
161     for(int i=0; i<SOCKET_PER_THREAD; i++)
162     {
163         maxfd = max(socket[i], maxfd);
164     }
165 
166     while(1)
167     {
168         fd_set recv_fds;
169         for(int i=0; i<SOCKET_PER_THREAD; i++)
170         {
171             FD_ZERO(&recv_fds);
172             FD_SET(socket[i], &recv_fds);
173         }
174         poll_and_recv(recv_fds, maxfd, socket);
175     }
176     return 0;
177 }
178 
179 
180 int main()
181 {
182     int sock[SERVER_NUM][SOCKNUM/SERVER_NUM];
183     struct sockaddr_in addr_ser[SERVER_NUM];
184     struct sockaddr_in addr_cli[SERVER_NUM][SOCKNUM/SERVER_NUM];
185 
186     std::string local_ip("192.168.250.141");
187 
188     std::string server_ip[SERVER_NUM] = {"192.168.250.146", "192.168.250.147", "192.168.250.142", "192.168.250.143"};
189     //std::string server_ip[SERVER_NUM] = {"192.168.251.166", "192.168.251.167", "192.168.251.162", "192.168.251.163"}; 
190 //  std::string server_ip[SERVER_NUM] = {"192.168.251.163"}; 
191     for(int ser=0; ser < SERVER_NUM; ser++)
192     {
193     for(int i=0; i<SOCKNUM/SERVER_NUM; i++)
194     {
195         sock[ser][i] = socket(AF_INET, SOCK_STREAM, 0);
196         if(sock[ser][i] < 0)
197         {
198             printf("%d ", i);
199             perror("create socket fail");
200         }
201 
202         addr_ser[ser].sin_family = AF_INET;
203         addr_ser[ser].sin_port = htons(PORT);
204         addr_ser[ser].sin_addr.s_addr = inet_addr(server_ip[ser].c_str());
205 
206         addr_cli[ser][i].sin_family = AF_INET;
207         addr_cli[ser][i].sin_port = 0;
208         addr_cli[ser][i].sin_addr.s_addr = inet_addr(local_ip.c_str());
209 
210 
211         int sockopt = 1;
212         if ( setsockopt(sock[ser][i], SOL_SOCKET, SO_REUSEADDR, (char*)&sockopt, sizeof(sockopt)) == -1 )
213         {
214             perror("set reuseaddr error");
215             exit(1);
216         }
217 
218 
219 #if 0 
220 
221         if ( SetSocketOptions(sock[ser][i]) == -1)
222         {
223             perror("set socket options error");
224             exit(1);
225         }
226 #endif
227 
228         if( bind(sock[ser][i], (struct sockaddr*)&addr_cli[ser][i], sizeof(addr_cli[ser][i]) ) < 0 )
229         {
230             perror("TCP bind: ");
231             exit(1);
232         }
233         printf("bind ok!\n");
234 
235         if(connect(sock[ser][i], (struct sockaddr*)&addr_ser[ser], sizeof(struct sockaddr)) < 0)
236         {
237             perror("connect fail:");
238             exit(1);
239         }
240         printf("connect ok!\n");
241 
242     }
243 
244     }
245 
246 
247     int socket[SOCKNUM] ;
248     int count=0;
249     for(int i=0; i< SERVER_NUM; i++)
250     {
251         for(int j=0; j<SOCKNUM/SERVER_NUM; j++)
252         {
253             socket[count++] = sock[i][j];
254         }
255     }
256 
257     pthread_t tid[THREAD_NUM];
258     ARG a[THREAD_NUM];
259     for(int i=0; i<THREAD_NUM; i++)
260     {
261         for(int j=0; j<SOCKET_PER_THREAD; j++)
262         {
263         a[i].sock[j] = socket[i*SOCKET_PER_THREAD+j];
264         }
265         pthread_create(&tid[i], 0, recvData, (void *)&a[i]);
266     }
267 
268     for(int i=0; i<SOCKNUM; i++)
269     {
270         pthread_join(tid[i], 0);
271     }
272 
273     return 0;
274 }

 

计算带宽代码：bw.sh

1 #!/bin/bash
2 
3 
4 bw1=`ifconfig eth4 |grep "RX bytes"|awk ‘{print $2}‘|awk -F : ‘{print $2}‘`
5 sleep 60
6 bw2=`ifconfig eth4 |grep "RX bytes"|awk ‘{print $2}‘|awk -F : ‘{print $2}‘`
7 
8 echo "($bw2-$bw1)/1024.0/1024.0/1024.0*8/60.0"|bc -l