From cb85aa8a8d02a3d6dc16e3f32e78da9e70f9c7f5 Mon Sep 17 00:00:00 2001 From: wangzhengquan <wangzhengquan85@126.com> Date: 星期二, 02 二月 2021 17:49:21 +0800 Subject: [PATCH] update --- test_net_socket/test_net_mod_socket.cpp | 214 +++++++++++++++++++++++++++++++++++++++++------------ 1 files changed, 164 insertions(+), 50 deletions(-) diff --git a/test_net_socket/test_net_mod_socket.cpp b/test_net_socket/test_net_mod_socket.cpp index a1a47ec..94be25b 100644 --- a/test_net_socket/test_net_mod_socket.cpp +++ b/test_net_socket/test_net_mod_socket.cpp @@ -1,3 +1,4 @@ +#include <assert.h> #include "net_mod_server_socket_wrapper.h" #include "net_mod_socket_wrapper.h" #include "bus_server_socket_wrapper.h" @@ -5,12 +6,14 @@ #include "shm_mm_wrapper.h" #include "usg_common.h" #include <getopt.h> +#include "logger_factory.h" #define SCALE 100000 typedef struct Targ { + net_node_t *node; char *nodelist; - int id; + long id; }Targ; @@ -27,7 +30,7 @@ argument_t parse_args (int argc, char *argv[]); void usage(char *name); -int parse_node_list(char *str, net_node_t *node_arr_addr[]) ; +int parse_node_list(const char *str, net_node_t *node_arr_addr[]) ; void print_node_list(net_node_t *node_arr, int len); @@ -78,7 +81,7 @@ int key; int rv; while ((rv = net_mod_socket_recvfrom( sockt, &recvbuf, &size, &key) ) == 0) { - printf("鏀跺埌璁㈤槄娑堟伅:%s\n", recvbuf); + printf("鏀跺埌璁㈤槄娑堟伅:%s\n", (char *)recvbuf); free(recvbuf); } @@ -132,17 +135,17 @@ void start_reply(int key) { printf("start reply\n"); - void *client = net_mod_socket_open(); - net_mod_socket_bind(client, key); + void *ser = net_mod_socket_open(); + net_mod_socket_bind(ser, key); int size; void *recvbuf; char sendbuf[512]; int rv; int remote_port; - while ( (rv = net_mod_socket_recvfrom(client, &recvbuf, &size, &remote_port) ) == 0) { + while ( (rv = net_mod_socket_recvfrom(ser, &recvbuf, &size, &remote_port) ) == 0) { // printf( "server: RECEIVED REQUEST FROM PORT %d NAME %s\n", remote_port, recvbuf); - sprintf(sendbuf, "RECEIVED PORT %d NAME %s", remote_port, recvbuf); - net_mod_socket_sendto(client, sendbuf, strlen(sendbuf) + 1, remote_port); + sprintf(sendbuf, "%d RECEIVED %s", net_mod_socket_get_key(ser), (char *)recvbuf); + net_mod_socket_sendto(ser, sendbuf, strlen(sendbuf) + 1, remote_port); free(recvbuf); } } @@ -194,11 +197,11 @@ n = net_mod_socket_sendandrecv_timeout(client, node_arr, node_arr_size, content, strlen(content), &recv_arr, &recv_arr_size, 1000); printf(" %d nodes reply\n", n); for(i=0; i<recv_arr_size; i++) { - printf("host:%s, port: %d, key:%d, content: %s\n", + printf("reply from (host:%s, port: %d, key:%d) >> %s\n", recv_arr[i].host, recv_arr[i].port, recv_arr[i].key, - recv_arr[i].content + (char *)recv_arr[i].content ); } @@ -243,71 +246,91 @@ } -void *_run_sendandrecv_(void *arg) { +void *_run_one_sendto_many_(void *arg) { Targ *targ = (Targ *)arg; - char sendbuf[512]; + char sendbuf[128]; - int i,j, n, recv_arr_size; + int j, n; + int recv_arr_size; net_mod_recv_msg_t *recv_arr; int total = 0; - void *client = net_mod_socket_open(); - net_node_t *node_arr; - int node_arr_size = parse_node_list(targ->nodelist, &node_arr); - + int rkey, lkey; + unsigned int l = 0 , rl; + const char *hello_format = "%d say Hello %d"; + const char *reply_format = "%d RECEIVED %d say Hello %d"; - char filename[512]; - sprintf(filename, "test%d.tmp", targ->id); + char filename[128]; + sprintf(filename, "test%d.tmp", targ->node->key); 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); + for (l = 0; l < SCALE; l++) { + sprintf(sendbuf, hello_format, net_mod_socket_get_key(client), l); + // 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); - n = net_mod_socket_sendandrecv_timeout(client, node_arr, node_arr_size, sendbuf, strlen(sendbuf) + 1, &recv_arr, &recv_arr_size, 1000); - printf("%d: send %d nodes\n", i, n); + n = net_mod_socket_sendandrecv_timeout(client, targ->node, 1, sendbuf, strlen(sendbuf) + 1, &recv_arr, &recv_arr_size, 1); + printf("%d: send %d nodes\n", l, 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 - ); + + fprintf(stdout, "%d send '%s' to %d. received from (host=%s, port= %d, key=%d) '%s'\n", + net_mod_socket_get_key(client), + sendbuf, + targ->node->key, + recv_arr[j].host, + recv_arr[j].port, + recv_arr[j].key, + (char *)recv_arr[j].content + ); + + printf("key == %d\n", net_mod_socket_get_key(client)); + assert(sscanf((const char *)recv_arr[j].content, reply_format, &rkey, &lkey, &rl) == 3); + assert(targ->node->key == rkey); + assert(net_mod_socket_get_key(client) == lkey); + assert(rl == l); } // 浣跨敤瀹屽悗锛屼笉瑕佸繕璁伴噴鏀炬帀 net_mod_socket_free_recv_msg_arr(recv_arr, recv_arr_size); total += n; } - fclose(fp); - net_mod_socket_close(client); + if(fp != NULL) + fclose(fp); + // net_mod_socket_close(client); return (void *)total; } -void test_net_sendandrecv_threads(char *nodelist) { +//澶氱嚎绋媠end +void one_sendto_many(char *nodelist) { - int status, i = 0, processors = 1; - void *res[processors]; + int status, i = 0; + // Targ *targs = (Targ *)calloc(processors, sizeof(Targ)); - Targ targs[processors]; - pthread_t tids[processors]; + char sendbuf[512]; struct timeval start, end; long total = 0; + client = net_mod_socket_open(); + net_mod_socket_bind(client, shm_mm_wrapper_alloc_key()); + net_node_t *node_arr; + int node_arr_size = parse_node_list(nodelist, &node_arr); + Targ targs[node_arr_size]; + pthread_t tids[node_arr_size]; + void *res[node_arr_size]; + printf("寮�濮嬫祴璇�...\n"); gettimeofday(&start, NULL); - for (i = 0; i < processors; i++) { - targs[i].nodelist = nodelist; + for (i = 0; i < node_arr_size; i++) { + targs[i].node = node_arr + i; targs[i].id = i; - pthread_create(&tids[i], NULL, _run_sendandrecv_, (void *)&targs[i]); + pthread_create(&tids[i], NULL, _run_one_sendto_many_, (void *)&targs[i]); } - for (i = 0; i < processors; i++) { + for (i = 0; i < node_arr_size; i++) { if (pthread_join(tids[i], &res[i]) != 0) { perror("multyThreadClient pthread_join"); } else { @@ -326,10 +349,63 @@ } +// 鏃犻檺寰幆send +void test_net_sendandrecv(char *nodelist) { + + int n, j; + void * client; + int recv_arr_size; + net_mod_recv_msg_t *recv_arr; + net_node_t *node_arr; + int node_arr_size = parse_node_list(nodelist, &node_arr); + char buf[128]; + pid_t pid, retPid ; + unsigned int l , retl; + int remoteKey; + const char *hello_format = "%d say Hello %u "; + const char *reply_format = "%d RECEIVED %d say Hello %d"; + + pid = getpid(); + l = 0; + + client = net_mod_socket_open(); + while(true) { + sprintf(buf, hello_format, pid, l); + n = net_mod_socket_sendandrecv_timeout(client, node_arr, node_arr_size, buf, strlen(buf)+1, + &recv_arr, &recv_arr_size, 1); + printf(" %d nodes reply\n", n); + for(j = 0; j < recv_arr_size; j++) { + + printf("%ld send '%s' . received '%s' from (host:%s, port: %d, key:%d) \n", + (long)pid, + buf, + (char *)recv_arr[j].content, + recv_arr[j].host, + recv_arr[j].port, + recv_arr[j].key + + ); + + + + assert(sscanf((const char *)recv_arr[j].content, reply_format, &remoteKey, &retPid, &retl) == 3); + assert(retPid == pid); + assert(retl == l); + assert(remoteKey == recv_arr[j].key); + } + + // 浣跨敤瀹屽悗锛屼笉瑕佸繕璁伴噴鏀炬帀 + net_mod_socket_free_recv_msg_arr(recv_arr, recv_arr_size); + l++; + } + + net_mod_socket_close(client); + +} void *_run_pub_(void *arg) { Targ *targ = (Targ *)arg; - char sendbuf[512]; + char sendbuf[128]; int i,j, n; int total = 0; @@ -338,9 +414,6 @@ int node_arr_size = parse_node_list(targ->nodelist, &node_arr); char *topic = "news"; - - - // char filename[512]; // sprintf(filename, "test%d.tmp", targ->id); // FILE *fp = NULL; @@ -349,11 +422,11 @@ for (i = 0; i < SCALE; i++) { - sprintf(sendbuf, "thread(%d) %d", targ->id, i); + sprintf(sendbuf, "thread(%ld) %d", targ->id, i); n = net_mod_socket_pub(client, node_arr, node_arr_size, topic, strlen(topic)+1, sendbuf, strlen(sendbuf)+1); // n = net_mod_socket_pub(client, node_arr, node_arr_size, sendbuf, strlen(sendbuf) + 1, &recv_arr, &recv_arr_size); - printf( "pub:%s to %d nodes\n", sendbuf, n); + LoggerFactory::getLogger()->debug( "pub:%s to %d nodes\n", sendbuf, n); total += n; } // fclose(fp); @@ -361,6 +434,7 @@ return (void *)total; } +//澶氱嚎绋媝ub void test_net_pub_threads(char *nodelist) { int status, i = 0, processors = 4; @@ -399,6 +473,28 @@ // fflush(stdout); net_mod_socket_close(client); } + +// 鏃犻檺寰幆pub +void test_net_pub(char *nodelist) { + + int n; + char sendbuf[512]; + net_node_t *node_arr; + int node_arr_size = parse_node_list(nodelist, &node_arr); + + char *topic = "news"; + sprintf(sendbuf, "pid:%ld: Good news!!", (long)getpid()); + + void * client = net_mod_socket_open(); + while (true) { + n = net_mod_socket_pub(client, node_arr, node_arr_size, topic, strlen(topic)+1, sendbuf, strlen(sendbuf)+1); + // n = net_mod_socket_pub(client, node_arr, node_arr_size, sendbuf, strlen(sendbuf) + 1, &recv_arr, &recv_arr_size); + LoggerFactory::getLogger()->debug( "pub to %d nodes\n", n); + } + net_mod_socket_close(client); +} + + @@ -447,14 +543,23 @@ } start_net_client(opt.sendlist, opt.publist); } - else if (strcmp("test_net_sendandrecv_threads", opt.fun) == 0) { + else if (strcmp("one_sendto_many", opt.fun) == 0) { if(opt.sendlist == 0) { fprintf(stderr, "Missing sendlist .\n"); usage(argv[0]); exit(1); } - test_net_sendandrecv_threads(opt.sendlist); + one_sendto_many(opt.sendlist); + } + else if (strcmp("test_net_sendandrecv", opt.fun) == 0) { + if(opt.sendlist == 0) { + fprintf(stderr, "Missing sendlist .\n"); + usage(argv[0]); + exit(1); + } + + test_net_sendandrecv(opt.sendlist); } else if (strcmp("test_net_pub_threads", opt.fun) == 0) { if(opt.publist == 0) { @@ -464,6 +569,15 @@ } test_net_pub_threads(opt.publist); + } + else if (strcmp("test_net_pub", opt.fun) == 0) { + if(opt.publist == 0) { + fprintf(stderr, "Missing publist .\n"); + usage(argv[0]); + exit(1); + } + + test_net_pub(opt.publist); } else { @@ -619,7 +733,7 @@ * } * @return 鏁扮粍鐨勯暱搴� */ -int parse_node_list(char *str, net_node_t *node_arr_addr[]) { +int parse_node_list(const char *str, net_node_t *node_arr_addr[]) { int i, j; char **property_arr; int property_arr_len; -- Gitblit v1.8.0