硬件：刀片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 }