diff --git a/src/common/src/tglobal.c b/src/common/src/tglobal.c index 0cd3196ad015e28e6444a5763fae6eb2c09b8d35..c20061d11b76c452b5c64dba337a59f3466b6759 100644 --- a/src/common/src/tglobal.c +++ b/src/common/src/tglobal.c @@ -140,7 +140,7 @@ char tsMqttBrokerAddress[128] = {0}; char tsMqttBrokerClientId[128] = {0}; // monitor -int32_t tsEnableMonitorModule = 0; +int32_t tsEnableMonitorModule = 1; char tsMonitorDbName[TSDB_DB_NAME_LEN] = "log"; char tsInternalPass[] = "secretkey"; int32_t tsMonitorInterval = 30; // seconds diff --git a/src/kit/CMakeLists.txt b/src/kit/CMakeLists.txt index df3ce1000197b1add7f6454de7ed13c652a4d50d..7095d797555091ac553ffa6e023ed47f75f9305d 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 0000000000000000000000000000000000000000..a7412b196d526294091f27eadc1200fd5ed10416 --- /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 706359ec20e5b014630164a110b8997d2e1d76fc..4b72c5c8598710501ced9e7c86cc51250a91e592 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 (64*1000) +#define BUFFER_SIZE (MAX_PKG_LEN + 1024) typedef struct { int port; - char *host[15]; -} info; + 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 64k 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,16 +63,21 @@ 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: + return ARGP_ERR_UNKNOWN; } return 0; } static struct argp argp = {options, parse_opt, 0, 0}; -void *checkPort(void *sarg) { - info *pinfo = (info *)sarg; - int port = pinfo->port; - char *host = *pinfo->host; +int checkTcpPort(info_s *info) { + int port = info->port; + char *host = info->host; int clientSocket; struct sockaddr_in serverAddr; @@ -76,39 +85,65 @@ void *checkPort(void *sarg) { char recvbuf[BUFFER_SIZE]; int iDataNum; if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) { - perror("socket"); - return NULL; + printf("socket() fail: %s\n", strerror(errno)); + return -1; } serverAddr.sin_family = AF_INET; serverAddr.sin_port = htons(port); serverAddr.sin_addr.s_addr = inet_addr(host); - printf("=================================\n"); + //printf("=================================\n"); if (connect(clientSocket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) { - perror("connect"); - return NULL; + 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, "client send tcp pkg to %s:%d, content: 1122334455", host, port); + sprintf(sendbuf + info->pktLen - 16, "1122334455667788"); + + send(clientSocket, sendbuf, info->pktLen, 0); + + memset(recvbuf, 0, BUFFER_SIZE); + int nleft, nread; + char *ptr = recvbuf; + nleft = info->pktLen; + while (nleft > 0) { + nread = recv(clientSocket, ptr, BUFFER_SIZE, 0);; + + if (nread == 0) { + break; + } else if (nread < 0) { + if (errno == EINTR) { + continue; + } else { + printf("recv ack pkg from TCP port: %d fail:%s.\n", port, strerror(errno)); + close(clientSocket); + return -1; + } + } else { + nleft -= nread; + ptr += nread; + iDataNum += nread; + } } - printf("Connect to: %s:%d...success\n", host, port); - - sprintf(sendbuf, "send port_%d", port); - send(clientSocket, sendbuf, strlen(sendbuf), 0); - printf("Send msg_%d: %s\n", port, sendbuf); - 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("recv 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 NULL; + return 0; } -void *checkUPort(void *sarg) { - info *pinfo = (info *)sarg; - int port = pinfo->port; - char *host = *pinfo->host; +int checkUdpPort(info_s *info) { + int port = info->port; + char *host = info->host; int clientSocket; struct sockaddr_in serverAddr; @@ -117,56 +152,76 @@ void *checkUPort(void *sarg) { int iDataNum; if ((clientSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { perror("socket"); - return NULL; + 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); - printf("=================================\n"); - - 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); - sendto(clientSocket, sendbuf, strlen(sendbuf), 0, (struct sockaddr *)&serverAddr, (int)sin_size); - - printf("Send msg_%d by udp: %s\n", port, sendbuf); + int code = sendto(clientSocket, sendbuf, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int)sin_size); + if (code < 0) { + perror("sendto"); + return -1; + } - 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); - printf("=================================\n"); + 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 NULL; + return 0; } int main(int argc, char *argv[]) { - SArguments arguments = {"127.0.0.1", 6041, 6050}; - + 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", 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; - char *host = arguments.host; - info *tinfo = malloc(sizeof(info)); - info *uinfo = malloc(sizeof(info)); - - for (; port < arguments.max_port; port++) { - printf("For test: %s:%d\n", host, port); - - *tinfo->host = host; - tinfo->port = port; - checkPort(tinfo); - *uinfo->host = host; - uinfo->port = port; - checkUPort(uinfo); + info.host = arguments.host; + info.pktLen = arguments.pktLen; + + 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.\t\t", port); + } else { + printf("tcp port:%d test ok.\t\t", port); + } + + ret = checkUdpPort(&info); + if (ret != 0) { + printf("udp port:%d test fail.\t\t", port); + } else { + printf("udp port:%d test ok.\t\t", port); + } } - free(tinfo); - free(uinfo); -} \ No newline at end of file + printf("\n"); + return 0; +} diff --git a/src/kit/taosnetwork/server.c b/src/kit/taosnetwork/server.c index c967828f0b28ea8bd55c91c3473a7762ac815356..1c3bc6fa098d793c2c7d1c355c524c755301b57b 100644 --- a/src/kit/taosnetwork/server.c +++ b/src/kit/taosnetwork/server.c @@ -27,24 +27,28 @@ #include #include #include +#include -#define BUFFER_SIZE 200 +#define MAX_PKG_LEN (64*1000) +#define BUFFER_SIZE (MAX_PKG_LEN + 1024) typedef struct { int port; - int type; // 0: tcp, 1: udo, default: 0 -} info; + 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 6020.", 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 6041.", 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 64k 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,16 +63,21 @@ 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: + return ARGP_ERR_UNKNOWN; } return 0; } static struct argp argp = {options, parse_opt, 0, 0}; -static void *bindPort(void *sarg) { - info *pinfo = (info *)sarg; +static void *bindTcpPort(void *sarg) { + info_s *pinfo = (info_s *)sarg; int port = pinfo->port; - int type = pinfo->type; int serverSocket; struct sockaddr_in server_addr; @@ -76,10 +85,10 @@ static void *bindPort(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; } @@ -89,65 +98,67 @@ static void *bindPort(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; } - printf("Bind port: %d success\n", port); + //printf("Bind port: %d success\n", port); 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); - - 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); + + memset(buffer, 0, BUFFER_SIZE); + int nleft, nread; + char *ptr = buffer; + nleft = pinfo->pktLen; + while (nleft > 0) { + nread = recv(client, ptr, BUFFER_SIZE, 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; } -static void *bindUPort(void *sarg) { - info *pinfo = (info *)sarg; +static void *bindUdpPort(void *sarg) { + info_s *pinfo = (info_s *)sarg; int port = pinfo->port; - int type = pinfo->type; int serverSocket; struct sockaddr_in server_addr; struct sockaddr_in clientAddr; - int addr_len = sizeof(clientAddr); - int client; char buffer[BUFFER_SIZE]; int iDataNum; - + if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) { perror("socket"); return NULL; @@ -164,10 +175,9 @@ static void *bindUPort(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); @@ -178,21 +188,10 @@ static void *bindUPort(void *sarg) { continue; } if (iDataNum > 0) { - printf("=================================\n"); - - 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'; + 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); - 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); - - send(client, buffer, strlen(buffer), 0); - printf("=================================\n"); + sendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int)sin_size); } } @@ -202,39 +201,44 @@ static void *bindUPort(void *sarg) { int main(int argc, char *argv[]) { - SArguments arguments = {"127.0.0.1", 6020, 6050}; + 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; + int num = arguments.max_port - arguments.port + 1; if (num < 0) { num = 1; } pthread_t *pids = malloc(2 * num * sizeof(pthread_t)); - info * infos = malloc(num * sizeof(info)); - info * uinfos = malloc(num * sizeof(info)); + info_s * tinfos = malloc(num * sizeof(info_s)); + info_s * uinfos = malloc(num * sizeof(info_s)); for (size_t i = 0; i < num; i++) { - info *pinfo = infos++; - pinfo->port = port; - - if (pthread_create(pids + i, NULL, bindPort, pinfo) != 0) //创建线程 - { //创建线程失败 - printf("创建线程失败: %d.\n", port); - exit(0); + info_s *tcpInfo = tinfos + i; + tcpInfo->port = port + i; + tcpInfo->pktLen = arguments.pktLen; + + if (pthread_create(pids + i, NULL, bindTcpPort, tcpInfo) != 0) + { + printf("create thread fail, port:%d.\n", port); + exit(-1); } - info *uinfo = uinfos++; - uinfo->port = port; - uinfo->type = 1; - port++; - if (pthread_create(pids + num + i, NULL, bindUPort, uinfo) != 0) //创建线程 - { //创建线程失败 - printf("创建线程失败: %d.\n", port); - exit(0); + info_s *udpInfo = uinfos + i; + udpInfo->port = port + i; + if (pthread_create(pids + num + i, NULL, bindUdpPort, udpInfo) != 0) + { + printf("create thread fail, port:%d.\n", port); + exit(-1); } } + for (int i = 0; i < num; i++) { pthread_join(pids[i], NULL); pthread_join(pids[(num + i)], NULL); diff --git a/src/mnode/src/mnodePeer.c b/src/mnode/src/mnodePeer.c index 71b8b1ea843dc799e18d38ac5fe5b464ab1170d7..7b4b4e4343c8fd7f4f2655162b25363da5c71394 100644 --- a/src/mnode/src/mnodePeer.c +++ b/src/mnode/src/mnodePeer.c @@ -75,6 +75,11 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg) { } void mnodeProcessPeerRsp(SRpcMsg *pMsg) { + if (!sdbIsMaster()) { + mError("%p, msg:%s is not processed for it is not master", pMsg->ahandle, taosMsg[pMsg->msgType]); + return; + } + if (tsMnodeProcessPeerRspFp[pMsg->msgType]) { (*tsMnodeProcessPeerRspFp[pMsg->msgType])(pMsg); } else { diff --git a/src/mnode/src/mnodeVgroup.c b/src/mnode/src/mnodeVgroup.c index e1c81db0ac83c03bf6f310c7191e0342c6a4e3db..ff253c1935376e337125ab8ae5e5ef702651c208 100644 --- a/src/mnode/src/mnodeVgroup.c +++ b/src/mnode/src/mnodeVgroup.c @@ -164,9 +164,9 @@ static int32_t mnodeVgroupActionUpdate(SSdbOper *pOper) { // reset vgid status on vgroup changed - mDebug("vgId:%d, reset sync status to unsynced", pVgroup->vgId); + mDebug("vgId:%d, reset sync status to offline", pVgroup->vgId); for (int32_t v = 0; v < pVgroup->numOfVnodes; ++v) { - pVgroup->vnodeGid[v].role = TAOS_SYNC_ROLE_UNSYNCED; + pVgroup->vnodeGid[v].role = TAOS_SYNC_ROLE_OFFLINE; } mnodeDecVgroupRef(pVgroup);