From ab5693295e2424d53b736248317823864dc94ce2 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 18 Aug 2020 02:11:33 +0000 Subject: [PATCH 1/3] TD-1165 --- src/balance/src/balance.c | 2 +- src/plugins/mqtt/src/mqttSystem.c | 6 +++++- src/util/src/tsocket.c | 8 ++++---- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/balance/src/balance.c b/src/balance/src/balance.c index 0d2d9fc778..2b1888042c 100644 --- a/src/balance/src/balance.c +++ b/src/balance/src/balance.c @@ -165,7 +165,7 @@ int32_t balanceAllocVnodes(SVgObj *pVgroup) { balanceSwapVnodeGid(pVgroup->vnodeGid, pVgroup->vnodeGid + 1); } } else { - int32_t randVal = rand() % 6; + int32_t randVal = randIndex++ % 6; if (randVal == 1) { // 1, 0, 2 balanceSwapVnodeGid(pVgroup->vnodeGid + 0, pVgroup->vnodeGid + 1); } else if (randVal == 2) { // 1, 2, 0 diff --git a/src/plugins/mqtt/src/mqttSystem.c b/src/plugins/mqtt/src/mqttSystem.c index ab3f3f5d84..b64c4715e8 100644 --- a/src/plugins/mqtt/src/mqttSystem.c +++ b/src/plugins/mqtt/src/mqttSystem.c @@ -47,7 +47,11 @@ int32_t mqttInitSystem() { recntStatus.recvbufsz = sizeof(recvbuf); char* url = tsMqttBrokerAddress; recntStatus.user_name = strstr(url, "@") != NULL ? strbetween(url, "//", ":") : NULL; - recntStatus.password = strstr(url, "@") != NULL ? strbetween(strstr(url, recntStatus.user_name), ":", "@") : NULL; + + char * passStr = strstr(url, recntStatus.user_name); + if (passStr != NULL) { + recntStatus.password = strstr(url, "@") != NULL ? strbetween(passStr, ":", "@") : NULL; + } if (strlen(url) == 0) { mqttDebug("mqtt module not init, url is null"); diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 23f2e5f00a..94e6d0e4e1 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -231,7 +231,7 @@ SOCKET taosOpenUdpSocket(uint32_t ip, uint16_t port) { localAddr.sin_addr.s_addr = ip; localAddr.sin_port = (uint16_t)htons(port); - if ((sockFd = (int)socket(AF_INET, SOCK_DGRAM, 0)) < 0) { + if ((sockFd = (int)socket(AF_INET, SOCK_DGRAM, 0)) <= 2) { uError("failed to open udp socket: %d (%s)", errno, strerror(errno)); return -1; } @@ -265,7 +265,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie sockFd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); - if (sockFd < 0) { + if (sockFd <= 2) { uError("failed to open the socket: %d (%s)", errno, strerror(errno)); return -1; } @@ -276,7 +276,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno)); taosCloseSocket(sockFd); return -1; - }; + } if (clientIp != 0) { memset((char *)&clientAddr, 0, sizeof(clientAddr)); @@ -371,7 +371,7 @@ SOCKET taosOpenTcpServerSocket(uint32_t ip, uint16_t port) { serverAdd.sin_addr.s_addr = ip; serverAdd.sin_port = (uint16_t)htons(port); - if ((sockFd = (int)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 2) { + if ((sockFd = (int)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) <= 2) { uError("failed to open TCP socket: %d (%s)", errno, strerror(errno)); return -1; } From 0698465236fb1f68e34f1e4033af939868fe2568 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 18 Aug 2020 06:40:33 +0000 Subject: [PATCH 2/3] The mqtt test has not passed, there are memory leaks, waiting for reconstruction --- src/plugins/mqtt/src/mqttSystem.c | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/plugins/mqtt/src/mqttSystem.c b/src/plugins/mqtt/src/mqttSystem.c index b64c4715e8..69810e2785 100644 --- a/src/plugins/mqtt/src/mqttSystem.c +++ b/src/plugins/mqtt/src/mqttSystem.c @@ -39,6 +39,7 @@ int mttIsRuning = 1; int32_t mqttInitSystem() { int rc = 0; +#if 0 uint8_t sendbuf[2048]; uint8_t recvbuf[1024]; recntStatus.sendbuf = sendbuf; @@ -95,11 +96,13 @@ int32_t mqttInitSystem() { topicPath = NULL; } +#endif return rc; } int32_t mqttStartSystem() { int rc = 0; +#if 0 if (recntStatus.user_name != NULL && recntStatus.password != NULL) { mqttInfo("connecting to mqtt://%s:%s@%s:%s/%s/", recntStatus.user_name, recntStatus.password, recntStatus.hostname, recntStatus.port, topicPath); @@ -116,18 +119,22 @@ int32_t mqttStartSystem() { } else { mqttInfo("listening for '%s' messages.", recntStatus.topic); } +#endif return rc; } void mqttStopSystem() { +#if 0 mqttClient.error = MQTT_ERROR_SOCKET_ERROR; mttIsRuning = 0; usleep(300000U); mqttCleanup(EXIT_SUCCESS, mqttClient.socketfd, &clientDaemonThread); mqttInfo("mqtt is stoped"); +#endif } void mqttCleanUpSystem() { +#if 0 mqttInfo("starting to cleanup mqtt"); free(recntStatus.user_name); free(recntStatus.password); @@ -136,6 +143,7 @@ void mqttCleanUpSystem() { free(recntStatus.topic); free(topicPath); mqttInfo("mqtt is cleaned up"); +#endif } void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published) { @@ -187,9 +195,11 @@ void* mqttClientRefresher(void* client) { } void mqttCleanup(int status, int sockfd, pthread_t* client_daemon) { +#if 0 mqttInfo("clean up mqtt module"); if (sockfd != -1) close(sockfd); if (client_daemon != NULL) pthread_cancel(*client_daemon); +#endif } void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code) { From 152867cb2ddf11444233f56846e8476fa027d36a Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 18 Aug 2020 06:49:24 +0000 Subject: [PATCH 3/3] minor changes --- src/util/src/tsocket.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 94e6d0e4e1..2de3490459 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -233,6 +233,7 @@ SOCKET taosOpenUdpSocket(uint32_t ip, uint16_t port) { if ((sockFd = (int)socket(AF_INET, SOCK_DGRAM, 0)) <= 2) { uError("failed to open udp socket: %d (%s)", errno, strerror(errno)); + close(sockFd); return -1; } @@ -267,6 +268,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie if (sockFd <= 2) { uError("failed to open the socket: %d (%s)", errno, strerror(errno)); + close(sockFd); return -1; } @@ -373,6 +375,7 @@ SOCKET taosOpenTcpServerSocket(uint32_t ip, uint16_t port) { if ((sockFd = (int)socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) <= 2) { uError("failed to open TCP socket: %d (%s)", errno, strerror(errno)); + close(sockFd); return -1; }