diff --git a/src/balance/src/balance.c b/src/balance/src/balance.c index 0d2d9fc778b437b8fe0671487da097f6c531b204..2b1888042c0645a93ae08fe9438085f96e547b71 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 ab3f3f5d841bbfd5c2c2f7cd439c445ccaae93b2..69810e27858040391e80fd39fea6af1ea8d4d96c 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; @@ -47,7 +48,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"); @@ -91,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); @@ -112,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); @@ -132,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) { @@ -183,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) { diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 23f2e5f00ae339937a9812ca0deab901275e6a89..2de3490459308d84aaea47f31a1c38adf7075b84 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -231,8 +231,9 @@ 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)); + close(sockFd); return -1; } @@ -265,8 +266,9 @@ 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)); + close(sockFd); return -1; } @@ -276,7 +278,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,8 +373,9 @@ 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)); + close(sockFd); return -1; }