Merge pull request #22372 from taosdata/fix/TD-25626

fix taosd cannot quit

include/libs/transport/trpc.h

@@ -89,7 +89,7 @@ typedef struct SRpcInit {
   int32_t retryMinInterval;  // retry init interval
   int32_t retryStepFactor;   // retry interval factor
   int32_t retryMaxInterval;  // retry max interval
-  int64_t retryMaxTimouet;
+  int64_t retryMaxTimeout;
 
   int32_t failFastThreshold;
   int32_t failFastInterval;

source/client/src/clientEnv.c

@@ -169,7 +169,7 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.retryMinInterval = tsRedirectPeriod;
   rpcInit.retryStepFactor = tsRedirectFactor;
   rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
-  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
+  rpcInit.retryMaxTimeout = tsMaxRetryWaitTime;
 
   int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
   connLimitNum = TMAX(connLimitNum, 10);

source/common/src/tglobal.c

@@ -47,7 +47,7 @@ bool    tsPrintAuth = false;
 
 // queue & threads
 int32_t tsNumOfRpcThreads = 1;
-int32_t tsNumOfRpcSessions = 10000;
+int32_t tsNumOfRpcSessions = 30000;
 int32_t tsTimeToGetAvailableConn = 500000;
 int32_t tsKeepAliveIdle = 60;
 
@@ -1281,9 +1281,9 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) {
         //        tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval;
         //      } else if (strcasecmp("smlBatchSize", name) == 0) {
         //        tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32;
-      } else if(strcasecmp("smlTsDefaultName", name) == 0) {
+      } else if (strcasecmp("smlTsDefaultName", name) == 0) {
         tstrncpy(tsSmlTsDefaultName, cfgGetItem(pCfg, "smlTsDefaultName")->str, TSDB_COL_NAME_LEN);
-      } else if(strcasecmp("smlDot2Underline", name) == 0) {
+      } else if (strcasecmp("smlDot2Underline", name) == 0) {
         tsSmlDot2Underline = cfgGetItem(pCfg, "smlDot2Underline")->bval;
       } else if (strcasecmp("shellActivityTimer", name) == 0) {
         tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -299,7 +299,7 @@ int32_t dmInitClient(SDnode *pDnode) {
   rpcInit.retryMinInterval = tsRedirectPeriod;
   rpcInit.retryStepFactor = tsRedirectFactor;
   rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
-  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
+  rpcInit.retryMaxTimeout = tsMaxRetryWaitTime;
 
   rpcInit.failFastInterval = 5000;  // interval threshold(ms)
   rpcInit.failFastThreshold = 3;    // failed threshold

source/libs/transport/inc/transportInt.h

@@ -46,14 +46,14 @@ typedef struct {
   int8_t   connType;
   char     label[TSDB_LABEL_LEN];
   char     user[TSDB_UNI_LEN];  // meter ID
-  int32_t compatibilityVer;
+  int32_t  compatibilityVer;
   int32_t  compressSize;  // -1: no compress, 0 : all data compressed, size: compress data if larger than size
   int8_t   encryption;    // encrypt or not
-  
+
   int32_t retryMinInterval;  // retry init interval
   int32_t retryStepFactor;   // retry interval factor
   int32_t retryMaxInterval;  // retry max interval
-  int32_t retryMaxTimouet;
+  int32_t retryMaxTimeout;
 
   int32_t failFastThreshold;
   int32_t failFastInterval;

source/libs/transport/src/trans.c

@@ -55,7 +55,7 @@ void* rpcOpen(const SRpcInit* pInit) {
   pRpc->retryMinInterval = pInit->retryMinInterval;  // retry init interval
   pRpc->retryStepFactor = pInit->retryStepFactor;
   pRpc->retryMaxInterval = pInit->retryMaxInterval;
-  pRpc->retryMaxTimouet = pInit->retryMaxTimouet;
+  pRpc->retryMaxTimeout = pInit->retryMaxTimeout;
 
   pRpc->failFastThreshold = pInit->failFastThreshold;
   pRpc->failFastInterval = pInit->failFastInterval;

source/libs/transport/src/transCli.c

@@ -2287,7 +2287,7 @@ bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
     pCtx->retryMinInterval = pTransInst->retryMinInterval;
     pCtx->retryMaxInterval = pTransInst->retryMaxInterval;
     pCtx->retryStepFactor = pTransInst->retryStepFactor;
-    pCtx->retryMaxTimeout = pTransInst->retryMaxTimouet;
+    pCtx->retryMaxTimeout = pTransInst->retryMaxTimeout;
     pCtx->retryInitTimestamp = taosGetTimestampMs();
     pCtx->retryNextInterval = pCtx->retryMinInterval;
     pCtx->retryStep = 0;

source/libs/transport/src/transSvr.c

@@ -726,7 +726,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
       tError("read error %s", uv_err_name(nread));
     }
     // TODO(log other failure reason)
-    tWarn("failed to create connect:%p", q);
+    tWarn("failed to create connect:%p, reason: %s", q, uv_err_name(nread));
     taosMemoryFree(buf->base);
     uv_close((uv_handle_t*)q, NULL);
     return;
@@ -741,10 +741,17 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
   uv_pipe_t* pipe = (uv_pipe_t*)q;
   if (!uv_pipe_pending_count(pipe)) {
     tError("No pending count");
+    uv_close((uv_handle_t*)q, NULL);
+    return;
+  }
+  if (pThrd->quit) {
+    tWarn("thread already received quit msg, ignore incoming conn");
+
+    uv_close((uv_handle_t*)q, NULL);
     return;
   }
 
-  uv_handle_type pending = uv_pipe_pending_type(pipe);
+  // uv_handle_type pending = uv_pipe_pending_type(pipe);
 
   SSvrConn* pConn = createConn(pThrd);