#include "net_mod_server_socket_wrapper.h" #include "net_mod_socket_wrapper.h" #include "shm_mm.h" #include "dgram_mod_socket.h" #include "usg_common.h" typedef struct Targ { int port; int id; }Targ; void server(int port) { void *serverSocket = net_mod_server_socket_open(port); if(net_mod_server_socket_start(serverSocket) != 0) { err_exit(errno, "net_mod_server_socket_start"); } } void client(int port ){ void * client = net_mod_socket_open(); char content[MAXLINE]; char action[512]; char topic[512]; int recv_arr_size, i, n; net_mod_recv_msg_t *recv_arr; //192.168.20.104 int node_arr_size = 2; net_node_t node_arr[] = { // {"192.168.5.22", port, 11}, {"192.168.20.10", port, 11}, {"192.168.20.104", port, 11} }; int pub_node_arr_size = 3; net_node_t pub_node_arr[] = { {"192.168.5.22", port, 8}, {"192.168.20.10", port, 8}, {"192.168.20.104", port, 8} }; while (true) { //printf("Usage: pub [content] or sub \n"); printf("Can I help you? pub, send or quit\n"); scanf("%s",action); if(strcmp(action, "pub") == 0) { printf("Please input topic and content\n"); scanf("%s %s", topic, content); n = net_mod_socket_pub(client, pub_node_arr, pub_node_arr_size, topic, strlen(topic)+1, content, strlen(content)+1); printf("pub %d nodes\n", n); } else if(strcmp(action, "send") == 0) { getc(stdin); printf("Please input content\n"); if (fgets(content, MAXLINE, stdin) != NULL) { // 收到消息的节点即使没有对应的信息, 也要回复一个表示无的消息,否则会一直等待 n = net_mod_socket_sendandrecv(client, node_arr, node_arr_size, content, strlen(content), &recv_arr, &recv_arr_size); printf("send %d nodes\n", n); for(i=0; iport; char sendbuf[512]; int i,j, n, recv_arr_size; net_mod_recv_msg_t *recv_arr; int node_arr_size = 2; //192.168.20.104 net_node_t node_arr[] = { // {"192.168.5.22", port, 11}, {"192.168.20.10", port, 11}, {"192.168.20.104", port, 11} }; void * client = net_mod_socket_open(); char filename[512]; sprintf(filename, "test%d.tmp", targ->id); FILE *fp = NULL; fp = fopen(filename, "w+"); // fp = stdout; int recvsize; void *recvbuf; for (i = 0; i < SCALE; i++) { sprintf(sendbuf, "thread(%d) %d", targ->id, i); fprintf(fp, "requst:%s\n", sendbuf); n = net_mod_socket_sendandrecv(client, node_arr, node_arr_size, sendbuf, strlen(sendbuf) + 1, &recv_arr, &recv_arr_size); //printf("send %d nodes\n", n); for(j=0; j < recv_arr_size; j++) { fprintf(fp, "reply: host:%s, port: %d, key:%d, content: %s\n", recv_arr[j].host, recv_arr[j].port, recv_arr[j].key, recv_arr[j].content ); } // 使用完后,不要忘记释放掉 net_mod_socket_free_recv_msg_arr(recv_arr, recv_arr_size); } fclose(fp); net_mod_socket_close(client); return (void *)i; } void mclient(int port) { int status, i = 0, processors = 2; void *res[processors]; // Targ *targs = (Targ *)calloc(processors, sizeof(Targ)); Targ targs[processors]; pthread_t tids[processors]; char sendbuf[512]; struct timeval start, end; long total = 0; printf("开始测试...\n"); gettimeofday(&start, NULL); for (i = 0; i < processors; i++) { targs[i].port = port; targs[i].id = i; pthread_create(&tids[i], NULL, runclient, (void *)&targs[i]); } for (i = 0; i < processors; i++) { if (pthread_join(tids[i], &res[i]) != 0) { perror("multyThreadClient pthread_join"); } else { total += (long)res[i]; //fprintf(stderr, "client(%d) 写入 %ld 条数据\n", i, (long)res[i]); } } gettimeofday(&end, NULL); double difftime = end.tv_sec * 1000000 + end.tv_usec - (start.tv_sec * 1000000 + start.tv_usec); long diffsec = (long) (difftime/1000000); long diffusec = difftime - diffsec*1000000; fprintf(stderr,"发送数目: %ld, 用时: (%ld sec %ld usec), 平均: %f\n", total, diffsec, diffusec, difftime/total ); // fflush(stdout); } int main(int argc, char *argv[]) { shm_init(512); int port; if (argc < 3) { fprintf(stderr, "Usage: %s %s|%s \n", argv[0], "server", "client"); return 1; } port = atoi(argv[2]); if (strcmp("server", argv[1]) == 0 ) { server(port); } if (strcmp("client", argv[1]) == 0) client(port); if (strcmp("mclient", argv[1]) == 0) mclient(port); }