From f2e9e5bf61b210d1c0bbb4989c16e9e97a46a016 Mon Sep 17 00:00:00 2001 From: Hui Li Date: Thu, 6 Aug 2020 10:47:34 +0800 Subject: [PATCH] [TD-1076] --- src/kit/CMakeLists.txt | 4 +- src/kit/taosnetwork/CMakeLists.txt | 10 +++ src/kit/taosnetwork/client.c | 86 +++++++++++++++++--------- src/kit/taosnetwork/server.c | 98 ++++++++++++++++-------------- 4 files changed, 124 insertions(+), 74 deletions(-) create mode 100644 src/kit/taosnetwork/CMakeLists.txt diff --git a/src/kit/CMakeLists.txt b/src/kit/CMakeLists.txt index df3ce10001..7095d79755 100644 --- a/src/kit/CMakeLists.txt +++ b/src/kit/CMakeLists.txt @@ -4,4 +4,6 @@ PROJECT(TDengine) ADD_SUBDIRECTORY(shell) ADD_SUBDIRECTORY(taosdemo) ADD_SUBDIRECTORY(taosdump) -ADD_SUBDIRECTORY(taosmigrate) \ No newline at end of file +ADD_SUBDIRECTORY(taosmigrate) +#ADD_SUBDIRECTORY(taosClusterTest) +ADD_SUBDIRECTORY(taosnetwork) diff --git a/src/kit/taosnetwork/CMakeLists.txt b/src/kit/taosnetwork/CMakeLists.txt new file mode 100644 index 0000000000..a7412b196d --- /dev/null +++ b/src/kit/taosnetwork/CMakeLists.txt @@ -0,0 +1,10 @@ +CMAKE_MINIMUM_REQUIRED(VERSION 2.8) +PROJECT(TDengine) + +IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM)) + AUX_SOURCE_DIRECTORY(. SRC) + ADD_EXECUTABLE(taosClient client.c) + ADD_EXECUTABLE(taosServer server.c) + TARGET_LINK_LIBRARIES( taosServer -lpthread -lm -lrt ) + TARGET_LINK_LIBRARIES( taosClient -lpthread -lm -lrt ) +ENDIF () diff --git a/src/kit/taosnetwork/client.c b/src/kit/taosnetwork/client.c index c51c1ab4ba..a4fc26f22f 100644 --- a/src/kit/taosnetwork/client.c +++ b/src/kit/taosnetwork/client.c @@ -28,23 +28,27 @@ #include #include -#define BUFFER_SIZE 200 +#define MAX_PKG_LEN (16*1000) +#define BUFFER_SIZE (MAX_PKG_LEN + 1024) typedef struct { int port; char *host; + uint16_t pktLen; } info_s; typedef struct Arguments { char * host; uint16_t port; uint16_t max_port; + uint16_t pktLen; } SArguments; static struct argp_option options[] = { {0, 'h', "host", 0, "The host to connect to TDEngine. Default is localhost.", 0}, - {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6041.", 1}, - {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2}}; + {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6030.", 1}, + {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6060.", 2}, + {0, 'l', "test pkg len", 0, "The len of pkg for test. Default is 1000 Bytes, max not greater than 16k Bytes.\nNotes: This parameter must be consistent between the client and the server.", 3}}; static error_t parse_opt(int key, char *arg, struct argp_state *state) { @@ -59,6 +63,12 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) { case 'm': arguments->max_port = atoi(arg); break; + case 'l': + arguments->pktLen = atoi(arg); + break; + default: + printf("unknow parameter!\n"); + break; } return 0; } @@ -75,7 +85,7 @@ int checkTcpPort(info_s *info) { char recvbuf[BUFFER_SIZE]; int iDataNum; if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) { - perror("socket"); + printf("socket() fail: %s\n", strerror(errno)); return -1; } serverAddr.sin_family = AF_INET; @@ -85,26 +95,30 @@ int checkTcpPort(info_s *info) { //printf("=================================\n"); if (connect(clientSocket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) { - perror("connect"); + printf("connect() fail: %s\n", strerror(errno)); return -1; } //printf("Connect to: %s:%d...success\n", host, port); + memset(sendbuf, 0, BUFFER_SIZE); + memset(recvbuf, 0, BUFFER_SIZE); - sprintf(sendbuf, "send port_%d", port); - send(clientSocket, sendbuf, strlen(sendbuf), 0); - //printf("Send msg_%d: %s\n", port, sendbuf); + sprintf(sendbuf, "client send tcp pkg to %s:%d, content: 1122334455", host, port); + sprintf(sendbuf + info->pktLen - 16, "1122334455667788"); + + send(clientSocket, sendbuf, info->pktLen, 0); - recvbuf[0] = '\0'; iDataNum = recv(clientSocket, recvbuf, BUFFER_SIZE, 0); - recvbuf[iDataNum] = '\0'; - //printf("Read ack msg_%d: %s\n", port, recvbuf); + if (iDataNum < info->pktLen) { + printf("Read ack pkg len: %d, less than req pkg len: %d from tcp port: %d\n", iDataNum, info->pktLen, port); + return -1; + } + //printf("Read ack pkg len:%d from tcp port: %d, buffer: %s %s\n", info->pktLen, port, recvbuf, recvbuf+iDataNum-8); - //printf("=================================\n"); close(clientSocket); return 0; } -void *checkUdpPort(info_s *info) { +int checkUdpPort(info_s *info) { int port = info->port; char *host = info->host; int clientSocket; @@ -117,60 +131,74 @@ void *checkUdpPort(info_s *info) { perror("socket"); return -1; } + serverAddr.sin_family = AF_INET; serverAddr.sin_port = htons(port); serverAddr.sin_addr.s_addr = inet_addr(host); + + memset(sendbuf, 0, BUFFER_SIZE); + memset(recvbuf, 0, BUFFER_SIZE); - sprintf(sendbuf, "send msg port_%d by udp", port); + sprintf(sendbuf, "client send udp pkg to %s:%d, content: 1122334455", host, port); + sprintf(sendbuf + info->pktLen - 16, "1122334455667788"); socklen_t sin_size = sizeof(*(struct sockaddr *)&serverAddr); - int code = sendto(clientSocket, sendbuf, strlen(sendbuf), 0, (struct sockaddr *)&serverAddr, (int)sin_size); + int code = sendto(clientSocket, sendbuf, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int)sin_size); if (code < 0) { perror("sendto"); return -1; } - //printf("Send msg_%d by udp: %s\n", port, sendbuf); - - recvbuf[0] = '\0'; iDataNum = recvfrom(clientSocket, recvbuf, BUFFER_SIZE, 0, (struct sockaddr *)&serverAddr, &sin_size); - recvbuf[iDataNum] = '\0'; - //printf("Read ack msg_%d from udp: %s\n", port, recvbuf); + if (iDataNum < info->pktLen) { + printf("Read ack pkg len: %d, less than req pkg len: %d from udp port: %d\n", iDataNum, info->pktLen, port); + return -1; + } + + //printf("Read ack pkg len:%d from udp port: %d, buffer: %s %s\n", info->pktLen, port, recvbuf, recvbuf+iDataNum-8); close(clientSocket); return 0; } int main(int argc, char *argv[]) { - SArguments arguments = {"127.0.0.1", 6030, 6060}; + SArguments arguments = {"127.0.0.1", 6030, 6060, 1000}; info_s info; int ret; argp_parse(&argp, argc, argv, 0, 0, &arguments); + if (arguments.pktLen > MAX_PKG_LEN) { + printf("test pkg len overflow: %d, max len not greater than %d bytes\n", arguments.pktLen, MAX_PKG_LEN); + exit(0); + } - printf("host: %s\tport: %d\tmax_port: %d\n\n", arguments.host, arguments.port, arguments.max_port); + printf("host: %s\tport: %d\tmax_port: %d\tpkgLen: %d\n", arguments.host, arguments.port, arguments.max_port, arguments.pktLen); int port = arguments.port; info.host = arguments.host; + info.pktLen = arguments.pktLen; - for (; port < arguments.max_port; port++) { - printf("test: %s:%d\n", info.host, port); + for (; port <= arguments.max_port; port++) { + //printf("test: %s:%d\n", info.host, port); + printf("\n"); info.port = port; ret = checkTcpPort(&info); if (ret != 0) { - printf("tcp port:%d test fail.", port); + printf("tcp port:%d test fail.\t\t", port); } else { - printf("tcp port:%d test ok.", port); + printf("tcp port:%d test ok.\t\t", port); } - checkUdpPort(&info); + ret = checkUdpPort(&info); if (ret != 0) { - printf("udp port:%d test fail.", port); + printf("udp port:%d test fail.\t\t", port); } else { - printf("udp port:%d test ok.", port); + printf("udp port:%d test ok.\t\t", port); } } + printf("\n"); + return 0; } diff --git a/src/kit/taosnetwork/server.c b/src/kit/taosnetwork/server.c index f77d23f0a7..18565edb3f 100644 --- a/src/kit/taosnetwork/server.c +++ b/src/kit/taosnetwork/server.c @@ -29,22 +29,26 @@ #include #include -#define BUFFER_SIZE 200 +#define MAX_PKG_LEN (16*1000) +#define BUFFER_SIZE (MAX_PKG_LEN + 1024) typedef struct { int port; + uint16_t pktLen; } info_s; typedef struct Arguments { char * host; uint16_t port; uint16_t max_port; + uint16_t pktLen; } SArguments; static struct argp_option options[] = { {0, 'h', "host", 0, "The host to connect to TDEngine. Default is localhost.", 0}, {0, 'p', "port", 0, "The TCP or UDP port number to use for the connection. Default is 6041.", 1}, - {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2}}; + {0, 'm', "max port", 0, "The max TCP or UDP port number to use for the connection. Default is 6050.", 2}, + {0, 'l', "test pkg len", 0, "The len of pkg for test. Default is 1000 Bytes, max not greater than 16k Bytes.\nNotes: This parameter must be consistent between the client and the server.", 3}}; static error_t parse_opt(int key, char *arg, struct argp_state *state) { @@ -59,6 +63,11 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) { case 'm': arguments->max_port = atoi(arg); break; + case 'l': + arguments->pktLen = atoi(arg); + break; + default: + break; } return 0; } @@ -75,10 +84,10 @@ static void *bindTcpPort(void *sarg) { int addr_len = sizeof(clientAddr); int client; char buffer[BUFFER_SIZE]; - int iDataNum; + int iDataNum = 0; if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) { - perror("socket"); + printf("socket() fail: %s", strerror(errno)); return NULL; } @@ -88,12 +97,12 @@ static void *bindTcpPort(void *sarg) { server_addr.sin_addr.s_addr = htonl(INADDR_ANY); if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) { - perror("connect"); + printf("port:%d bind() fail: %s", port, strerror(errno)); return NULL; } if (listen(serverSocket, 5) < 0) { - perror("listen"); + printf("listen() fail: %s", strerror(errno)); return NULL; } @@ -101,34 +110,39 @@ static void *bindTcpPort(void *sarg) { while (1) { client = accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len); if (client < 0) { - perror("accept"); + printf("accept() fail: %s", strerror(errno)); continue; } - //printf("=================================\n"); - printf("Client ip is %s, Server port is %d\n", inet_ntoa(clientAddr.sin_addr), port); - while (1) { - buffer[0] = '\0'; - iDataNum = recv(client, buffer, BUFFER_SIZE, 0); + memset(buffer, 0, BUFFER_SIZE); + int nleft, nread; + char *ptr = buffer; + nleft = pinfo->pktLen; + while (nleft > 0) { + nread = recv(client, buffer, BUFFER_SIZE, 0); - if (iDataNum < 0) { - perror("recv null"); - continue; - } - if (iDataNum > 0) { - buffer[iDataNum] = '\0'; - //printf("read msg:%s\n", buffer); - if (strcmp(buffer, "quit") == 0) break; - buffer[0] = '\0'; - - sprintf(buffer, "ack port_%d", port); - //printf("send ack msg:%s\n", buffer); - - send(client, buffer, strlen(buffer), 0); + if (nread == 0) { break; - } + } else if (nread < 0) { + if (errno == EINTR) { + continue; + } else { + printf("recv Client: %s pkg from TCP port: %d fail:%s.\n", inet_ntoa(clientAddr.sin_addr), port, strerror(errno)); + close(serverSocket); + return NULL; + } + } else { + nleft -= nread; + ptr += nread; + iDataNum += nread; + } + } + + printf("recv Client: %s pkg from TCP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum); + if (iDataNum > 0) { + send(client, buffer, iDataNum, 0); + break; } - //printf("=================================\n"); } close(serverSocket); return NULL; @@ -143,7 +157,7 @@ static void *bindUdpPort(void *sarg) { struct sockaddr_in clientAddr; char buffer[BUFFER_SIZE]; int iDataNum; - + if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { perror("socket"); return NULL; @@ -160,10 +174,9 @@ static void *bindUdpPort(void *sarg) { } socklen_t sin_size; - //printf("Bind port: %d success\n", port); while (1) { - buffer[0] = '\0'; + memset(buffer, 0, BUFFER_SIZE); sin_size = sizeof(*(struct sockaddr *)&server_addr); @@ -174,19 +187,10 @@ static void *bindUdpPort(void *sarg) { continue; } if (iDataNum > 0) { - //printf("=================================\n"); + printf("recv Client: %s pkg from UDP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum); + //printf("Read msg from udp:%s ... %s\n", buffer, buffer+iDataNum-16); - printf("Client ip is %s, Server port is %d\n", inet_ntoa(clientAddr.sin_addr), port); - buffer[iDataNum] = '\0'; - //printf("Read msg from udp:%s\n", buffer); - if (strcmp(buffer, "quit") == 0) break; - buffer[0] = '\0'; - - sprintf(buffer, "ack port_%d by udp", port); - //printf("Send ack msg by udp:%s\n", buffer); - - sendto(serverSocket, buffer, strlen(buffer), 0, (struct sockaddr *)&clientAddr, (int)sin_size); - //printf("=================================\n"); + sendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int)sin_size); } } @@ -196,8 +200,13 @@ static void *bindUdpPort(void *sarg) { int main(int argc, char *argv[]) { - SArguments arguments = {"127.0.0.1", 6030, 6060}; + SArguments arguments = {"127.0.0.1", 6030, 6060, 1000}; argp_parse(&argp, argc, argv, 0, 0, &arguments); + if (arguments.pktLen > MAX_PKG_LEN) { + printf("test pkg len overflow: %d, max len not greater than %d bytes\n", arguments.pktLen, MAX_PKG_LEN); + exit(0); + } + int port = arguments.port; int num = arguments.max_port - arguments.port + 1; @@ -212,6 +221,7 @@ int main(int argc, char *argv[]) { for (size_t i = 0; i < num; i++) { info_s *tcpInfo = tinfos + i; tcpInfo->port = port + i; + tcpInfo->pktLen = arguments.pktLen; if (pthread_create(pids + i, NULL, bindTcpPort, tcpInfo) != 0) {