Merge pull request #24409 from taosdata/opt/optMsgOnMnode

opt msg on mnd

include/common/tmsgcb.h

@@ -53,9 +53,12 @@ typedef struct {
   void*                   mgmt;
   void*                   clientRpc;
   void*                   serverRpc;
+  void*                   statusRpc;
+  void*                   syncRpc;
   PutToQueueFp            putToQueueFp;
   GetQueueSizeFp          qsizeFp;
   SendReqFp               sendReqFp;
+  SendReqFp               sendSyncReqFp;
   SendRspFp               sendRspFp;
   RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
   ReleaseHandleFp         releaseHandleFp;
@@ -67,6 +70,7 @@ void    tmsgSetDefault(const SMsgCb* msgcb);
 int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg);
 int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype);
 int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg);
+int32_t tmsgSendSyncReq(const SEpSet* epSet, SRpcMsg* pMsg);
 void    tmsgSendRsp(SRpcMsg* pMsg);
 void    tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg);
 void    tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type);

include/util/tdef.h

@@ -260,7 +260,7 @@ typedef enum ELogicConditionType {
 #define TSDB_PASSWORD_LEN      32
 #define TSDB_USET_PASSWORD_LEN 129
 #define TSDB_VERSION_LEN       32
-#define TSDB_LABEL_LEN         12
+#define TSDB_LABEL_LEN         16
 #define TSDB_JOB_STATUS_LEN    32
 
 #define TSDB_CLUSTER_ID_LEN       40

source/common/src/tglobal.c

@@ -109,7 +109,7 @@ bool    tsEnableTelem = true;
 int32_t  tsTelemInterval = 43200;
 char     tsTelemServer[TSDB_FQDN_LEN] = "telemetry.tdengine.com";
 uint16_t tsTelemPort = 80;
-char *   tsTelemUri = "/report";
+char    *tsTelemUri = "/report";
 
 #ifdef TD_ENTERPRISE
 bool tsEnableCrashReport = false;
@@ -253,7 +253,7 @@ int32_t tsCompactPullupInterval = 10;
 int32_t tsMqRebalanceInterval = 2;
 int32_t tsStreamCheckpointInterval = 60;
 float   tsSinkDataRate = 2.0;
-int32_t tsStreamNodeCheckInterval = 15;
+int32_t tsStreamNodeCheckInterval = 16;
 int32_t tsTtlUnit = 86400;
 int32_t tsTtlPushIntervalSec = 10;
 int32_t tsTrimVDbIntervalSec = 60 * 60;  // interval of trimming db in all vgroups
@@ -695,8 +695,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
 
   if (cfgAddBool(pCfg, "audit", tsEnableAudit, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1;
   if (cfgAddBool(pCfg, "auditCreateTable", tsEnableAuditCreateTable, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
-  if (cfgAddInt32(pCfg, "auditInterval", tsAuditInterval, 500, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0)
+  if (cfgAddInt32(pCfg, "auditInterval", tsAuditInterval, 500, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
-    return -1;
 
   if (cfgAddBool(pCfg, "crashReporting", tsEnableCrashReport, CFG_SCOPE_BOTH, CFG_DYN_NONE) != 0) return -1;
   if (cfgAddBool(pCfg, "telemetryReporting", tsEnableTelem, CFG_SCOPE_BOTH, CFG_DYN_ENT_SERVER) != 0) return -1;
@@ -715,8 +714,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
       0)
     return -1;
-  if (cfgAddInt32(pCfg, "compactPullupInterval", tsCompactPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
+  if (cfgAddInt32(pCfg, "compactPullupInterval", tsCompactPullupInterval, 1, 10000, CFG_SCOPE_SERVER,
-      0)
+                  CFG_DYN_ENT_SERVER) != 0)
     return -1;
   if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
       0)
@@ -1382,7 +1381,7 @@ void taosCleanupCfg() {
 
 typedef struct {
   const char *optionName;
-  void *      optionVar;
+  void       *optionVar;
 } OptionNameAndVar;
 
 static int32_t taosCfgSetOption(OptionNameAndVar *pOptions, int32_t optionSize, SConfigItem *pItem, bool isDebugflag) {
@@ -1395,7 +1394,7 @@ static int32_t taosCfgSetOption(OptionNameAndVar *pOptions, int32_t optionSize,
     switch (pItem->dtype) {
       case CFG_DTYPE_BOOL: {
         int32_t flag = pItem->i32;
-        bool *  pVar = pOptions[d].optionVar;
+        bool   *pVar = pOptions[d].optionVar;
         uInfo("%s set from %d to %d", optName, *pVar, flag);
         *pVar = flag;
         terrno = TSDB_CODE_SUCCESS;
@@ -1470,8 +1469,7 @@ static int32_t taosCfgDynamicOptionsForServer(SConfig *pCfg, char *name) {
         {"stDebugFlag", &stDebugFlag},   {"sndDebugFlag", &sndDebugFlag},
     };
 
-    static OptionNameAndVar options[] = {
+    static OptionNameAndVar options[] = {{"audit", &tsEnableAudit},
-        {"audit", &tsEnableAudit},
                                          {"asynclog", &tsAsyncLog},
                                          {"disableStream", &tsDisableStream},
                                          {"enableWhiteList", &tsEnableWhiteList},
@@ -1502,8 +1500,7 @@ static int32_t taosCfgDynamicOptionsForServer(SConfig *pCfg, char *name) {
                                          {"s3PageCacheSize", &tsS3PageCacheSize},
                                          {"s3UploadDelaySec", &tsS3UploadDelaySec},
                                          {"supportVnodes", &tsNumOfSupportVnodes},
-        {"experimental", &tsExperimental}
+                                         {"experimental", &tsExperimental}};
-    };
 
     if (taosCfgSetOption(debugOptions, tListLen(debugOptions), pItem, true) != 0) {
       taosCfgSetOption(options, tListLen(options), pItem, false);
@@ -1700,8 +1697,7 @@ static int32_t taosCfgDynamicOptionsForClient(SConfig *pCfg, char *name) {
         {"uDebugFlag", &uDebugFlag},     {"simDebugFlag", &simDebugFlag},
     };
 
-    static OptionNameAndVar options[] = {
+    static OptionNameAndVar options[] = {{"asyncLog", &tsAsyncLog},
-        {"asyncLog", &tsAsyncLog},
                                          {"assert", &tsAssert},
                                          {"compressMsgSize", &tsCompressMsgSize},
                                          {"countAlwaysReturnValue", &tsCountAlwaysReturnValue},
@@ -1725,8 +1721,7 @@ static int32_t taosCfgDynamicOptionsForClient(SConfig *pCfg, char *name) {
                                          {"shellActivityTimer", &tsShellActivityTimer},
                                          {"slowLogThreshold", &tsSlowLogThreshold},
                                          {"useAdapter", &tsUseAdapter},
-        {"experimental", &tsExperimental}
+                                         {"experimental", &tsExperimental}};
-    };
 
     if (taosCfgSetOption(debugOptions, tListLen(debugOptions), pItem, true) != 0) {
       taosCfgSetOption(options, tListLen(options), pItem, false);

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -45,7 +45,7 @@ static void dmMayShouldUpdateIpWhiteList(SDnodeMgmt *pMgmt, int64_t ver) {
 
   SRetrieveIpWhiteReq req = {.ipWhiteVer = oldVer};
   int32_t             contLen = tSerializeRetrieveIpWhite(NULL, 0, &req);
-  void               *pHead = rpcMallocCont(contLen);
+  void *              pHead = rpcMallocCont(contLen);
   tSerializeRetrieveIpWhite(pHead, contLen, &req);
 
   SRpcMsg rpcMsg = {.pCont = pHead,
@@ -144,7 +144,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   req.ipWhiteVer = pMgmt->pData->ipWhiteVer;
 
   int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
-  void   *pHead = rpcMallocCont(contLen);
+  void *  pHead = rpcMallocCont(contLen);
   tSerializeSStatusReq(pHead, contLen, &req);
   tFreeSStatusReq(&req);
 
@@ -161,7 +161,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   SEpSet epSet = {0};
   int8_t epUpdated = 0;
   dmGetMnodeEpSet(pMgmt->pData, &epSet);
-  rpcSendRecvWithTimeout(pMgmt->msgCb.clientRpc, &epSet, &rpcMsg, &rpcRsp, &epUpdated, 5000);
+  rpcSendRecvWithTimeout(pMgmt->msgCb.statusRpc, &epSet, &rpcMsg, &rpcRsp, &epUpdated, tsStatusInterval * 5 * 1000);
   if (rpcRsp.code != 0) {
     dmRotateMnodeEpSet(pMgmt->pData);
     char tbuf[512];
@@ -189,7 +189,7 @@ void dmSendNotifyReq(SDnodeMgmt *pMgmt) {
   req.pVloads = vinfo.pVloads;
 
   int32_t contLen = tSerializeSNotifyReq(NULL, 0, &req);
-  void   *pHead = rpcMallocCont(contLen);
+  void *  pHead = rpcMallocCont(contLen);
   tSerializeSNotifyReq(pHead, contLen, &req);
   tFreeSNotifyReq(&req);
 
@@ -284,7 +284,7 @@ int32_t dmProcessServerRunStatus(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 }
 
 SSDataBlock *dmBuildVariablesBlock(void) {
-  SSDataBlock         *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  SSDataBlock *        pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
   size_t               size = 0;
   const SSysTableMeta *pMeta = NULL;
   getInfosDbMeta(&pMeta, &size);

source/dnode/mgmt/mgmt_dnode/src/dmWorker.c

@@ -47,6 +47,7 @@ static void *dmStatusThreadFp(void *param) {
 }
 
 SDmNotifyHandle dmNotifyHdl = {.state = 0};
+
 static void *dmNotifyThreadFp(void *param) {
   SDnodeMgmt *pMgmt = param;
   setThreadName("dnode-notify");

source/dnode/mgmt/node_mgmt/inc/dmMgmt.h

@@ -48,6 +48,8 @@ typedef struct {
 typedef struct {
   void        *serverRpc;
   void        *clientRpc;
+  void        *statusRpc;
+  void        *syncRpc;
   SDnodeHandle msgHandles[TDMT_MAX];
 } SDnodeTrans;
 
@@ -136,8 +138,10 @@ int32_t dmInitServer(SDnode *pDnode);
 void    dmCleanupServer(SDnode *pDnode);
 int32_t dmInitClient(SDnode *pDnode);
 int32_t dmInitStatusClient(SDnode *pDnode);
+int32_t dmInitSyncClient(SDnode *pDnode);
 void    dmCleanupClient(SDnode *pDnode);
 void    dmCleanupStatusClient(SDnode *pDnode);
+void    dmCleanupSyncClient(SDnode *pDnode);
 SMsgCb  dmGetMsgcb(SDnode *pDnode);
 #ifdef TD_MODULE_OPTIMIZE
 int32_t dmInitMsgHandle(SDnode *pDnode, SMgmtWrapper *wrappers);

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

@@ -94,6 +94,9 @@ int32_t dmInitDnode(SDnode *pDnode) {
   indexInit(tsNumOfCommitThreads);
   streamMetaInit();
 
+  dmInitStatusClient(pDnode);
+  dmInitSyncClient(pDnode);  
+
   dmReportStartup("dnode-transport", "initialized");
   dDebug("dnode is created, ptr:%p", pDnode);
   code = 0;
@@ -115,7 +118,9 @@ void dmCleanupDnode(SDnode *pDnode) {
 
   dmCleanupClient(pDnode);
   dmCleanupStatusClient(pDnode);
+  dmCleanupSyncClient(pDnode);
   dmCleanupServer(pDnode);
+
   dmClearVars(pDnode);
   rpcCleanup();
   streamMetaCleanup();

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

@@ -322,6 +322,23 @@ static inline int32_t dmSendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
     return 0;
   }
 }
+static inline int32_t dmSendSyncReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
+  SDnode *pDnode = dmInstance();
+  if (pDnode->status != DND_STAT_RUNNING && pMsg->msgType < TDMT_SYNC_MSG) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+    if (pDnode->status == DND_STAT_INIT) {
+      terrno = TSDB_CODE_APP_IS_STARTING;
+    } else {
+      terrno = TSDB_CODE_APP_IS_STOPPING;
+    }
+    dError("failed to send rpc msg:%s since %s, handle:%p", TMSG_INFO(pMsg->msgType), terrstr(), pMsg->info.handle);
+    return -1;
+  } else {
+    rpcSendRequest(pDnode->trans.syncRpc, pEpSet, pMsg, NULL);
+    return 0;
+  }
+}
 
 static inline void dmRegisterBrokenLinkArg(SRpcMsg *pMsg) { rpcRegisterBrokenLinkArg(pMsg); }
 
@@ -346,8 +363,8 @@ int32_t dmInitClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
 
   SRpcInit rpcInit = {0};
-  rpcInit.label = "DND-C";
+  rpcInit.label = "DNODE-CLI";
-  rpcInit.numOfThreads = tsNumOfRpcThreads;
+  rpcInit.numOfThreads = tsNumOfRpcThreads / 2;
   rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
@@ -366,7 +383,7 @@ int32_t dmInitClient(SDnode *pDnode) {
   rpcInit.failFastThreshold = 3;    // failed threshold
   rpcInit.ffp = dmFailFastFp;
 
-  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
+  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3) / 2;
   connLimitNum = TMAX(connLimitNum, 10);
   connLimitNum = TMIN(connLimitNum, 500);
 
@@ -390,7 +407,7 @@ int32_t dmInitStatusClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
 
   SRpcInit rpcInit = {0};
-  rpcInit.label = "DND-STATUS";
+  rpcInit.label = "DNODE-STA-CLI";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
   rpcInit.sessions = 1024;
@@ -421,16 +438,61 @@ int32_t dmInitStatusClient(SDnode *pDnode) {
   rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
   taosVersionStrToInt(version, &(rpcInit.compatibilityVer));
 
-  // pTrans->statusClientRpc = rpcOpen(&rpcInit);
+  pTrans->statusRpc = rpcOpen(&rpcInit);
-  // if (pTrans->statusClientRpc == NULL) {
+  if (pTrans->statusRpc == NULL) {
-  //   dError("failed to init dnode rpc status client");
+    dError("failed to init dnode rpc status client");
-  //   return -1;
+    return -1;
-  // }
+  }
 
   dDebug("dnode rpc status client is initialized");
   return 0;
 }
 
+int32_t dmInitSyncClient(SDnode *pDnode) {
+  SDnodeTrans *pTrans = &pDnode->trans;
+
+  SRpcInit rpcInit = {0};
+  rpcInit.label = "DNODE-SYNC-CLI";
+  rpcInit.numOfThreads = tsNumOfRpcThreads / 2;
+  rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
+  rpcInit.sessions = 1024;
+  rpcInit.connType = TAOS_CONN_CLIENT;
+  rpcInit.user = TSDB_DEFAULT_USER;
+  rpcInit.idleTime = tsShellActivityTimer * 1000;
+  rpcInit.parent = pDnode;
+  rpcInit.rfp = rpcRfp;
+  rpcInit.compressSize = tsCompressMsgSize;
+
+  rpcInit.retryMinInterval = tsRedirectPeriod;
+  rpcInit.retryStepFactor = tsRedirectFactor;
+  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
+  rpcInit.retryMaxTimeout = tsMaxRetryWaitTime;
+
+  rpcInit.failFastInterval = 5000;  // interval threshold(ms)
+  rpcInit.failFastThreshold = 3;    // failed threshold
+  rpcInit.ffp = dmFailFastFp;
+
+  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3) / 2;
+  connLimitNum = TMAX(connLimitNum, 10);
+  connLimitNum = TMIN(connLimitNum, 500);
+
+  rpcInit.connLimitNum = connLimitNum;
+  rpcInit.connLimitLock = 1;
+  rpcInit.supportBatch = 1;
+  rpcInit.batchSize = 8 * 1024;
+  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
+  taosVersionStrToInt(version, &(rpcInit.compatibilityVer));
+
+  pTrans->syncRpc = rpcOpen(&rpcInit);
+  if (pTrans->syncRpc == NULL) {
+    dError("failed to init dnode rpc sync client");
+    return -1;
+  }
+
+  dDebug("dnode rpc sync client is initialized");
+  return 0;
+}
+
 void dmCleanupClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
   if (pTrans->clientRpc) {
@@ -441,11 +503,19 @@ void dmCleanupClient(SDnode *pDnode) {
 }
 void dmCleanupStatusClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
-  // if (pTrans->statusClientRpc) {
+  if (pTrans->statusRpc) {
-  //   rpcClose(pTrans->statusClientRpc);
+    rpcClose(pTrans->statusRpc);
-  //   pTrans->statusClientRpc = NULL;
+    pTrans->statusRpc = NULL;
-  //   dDebug("dnode rpc status client is closed");
+    dDebug("dnode rpc status client is closed");
-  // }
+  }
+}
+void dmCleanupSyncClient(SDnode *pDnode) {
+  SDnodeTrans *pTrans = &pDnode->trans;
+  if (pTrans->syncRpc) {
+    rpcClose(pTrans->syncRpc);
+    pTrans->syncRpc = NULL;
+    dDebug("dnode rpc sync client is closed");
+  }
 }
 
 int32_t dmInitServer(SDnode *pDnode) {
@@ -486,7 +556,10 @@ SMsgCb dmGetMsgcb(SDnode *pDnode) {
   SMsgCb msgCb = {
       .clientRpc = pDnode->trans.clientRpc,
       .serverRpc = pDnode->trans.serverRpc,
+      .statusRpc = pDnode->trans.statusRpc,
+      .syncRpc = pDnode->trans.syncRpc,
       .sendReqFp = dmSendReq,
+      .sendSyncReqFp = dmSendSyncReq,
       .sendRspFp = dmSendRsp,
       .registerBrokenLinkArgFp = dmRegisterBrokenLinkArg,
       .releaseHandleFp = dmReleaseHandle,

source/dnode/mnode/impl/src/mndDump.c

@@ -32,6 +32,10 @@ int32_t sendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
   terrno = TSDB_CODE_INVALID_PTR;
   return -1;
 }
+int32_t sendSyncReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
+  terrno = TSDB_CODE_INVALID_PTR;
+  return -1;
+}
 
 char *i642str(int64_t val) {
   static char str[24] = {0};
@@ -568,6 +572,7 @@ void mndDumpSdb() {
   SMsgCb msgCb = {0};
   msgCb.reportStartupFp = reportStartup;
   msgCb.sendReqFp = sendReq;
+  msgCb.sendSyncReqFp = sendSyncReq;
   msgCb.sendRspFp = sendRsp;
   msgCb.mgmt = (SMgmtWrapper *)(&msgCb);  // hack
   tmsgSetDefault(&msgCb);
@@ -590,7 +595,7 @@ void mndDumpSdb() {
   dumpTopic(pSdb, json);
   dumpConsumer(pSdb, json);
   dumpSubscribe(pSdb, json);
-//  dumpOffset(pSdb, json);
+  //  dumpOffset(pSdb, json);
   dumpStream(pSdb, json);
   dumpAcct(pSdb, json);
   dumpAuth(pSdb, json);
@@ -605,7 +610,7 @@ void mndDumpSdb() {
   char     *pCont = tjsonToString(json);
   int32_t   contLen = strlen(pCont);
   char      file[] = "sdb.json";
-  TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC| TD_FILE_WRITE_THROUGH);
+  TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     mError("failed to write %s since %s", file, terrstr());

source/dnode/mnode/impl/src/mndMain.c

@@ -299,14 +299,14 @@ static bool mnodeIsNotLeader(SMnode *pMnode) {
 }
 
 static int32_t minCronTime() {
-  int64_t min = INT64_MAX;
+  int32_t min = INT32_MAX;
   min = TMIN(min, tsTtlPushIntervalSec);
   min = TMIN(min, tsTrimVDbIntervalSec);
   min = TMIN(min, tsTransPullupInterval);
   min = TMIN(min, tsCompactPullupInterval);
   min = TMIN(min, tsMqRebalanceInterval);
   min = TMIN(min, tsStreamCheckpointInterval);
-  min = TMIN(min, 5);  // checkpointRemain
+  min = TMIN(min, 6);  // checkpointRemain
   min = TMIN(min, tsStreamNodeCheckInterval);
 
   int64_t telemInt = TMIN(60, (tsTelemInterval - 1));
@@ -386,7 +386,8 @@ static void *mndThreadFp(void *param) {
     int64_t minCron = minCronTime();
     if (sec % minCron == 0 && mnodeIsNotLeader(pMnode)) {
       // not leader, do nothing
-      mTrace("timer not process since mnode is not leader, reason: %s", tstrerror(terrno)) terrno = 0;
+      mTrace("timer not process since mnode is not leader, reason: %s", tstrerror(terrno));
+      terrno = 0;
       continue;
     }
     mndDoTimerPullupTask(pMnode, sec);

source/dnode/mnode/impl/test/trans/trans2.cpp

@@ -61,6 +61,7 @@ class MndTestTrans2 : public ::testing::Test {
     static SMsgCb msgCb = {0};
     msgCb.reportStartupFp = reportStartup;
     msgCb.sendReqFp = sendReq;
+    msgCb.sendSyncReqFp = sendSyncReq;
     msgCb.sendRspFp = sendRsp;
     msgCb.queueFps[SYNC_QUEUE] = putToQueue;
     msgCb.queueFps[WRITE_QUEUE] = putToQueue;

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -14,11 +14,11 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "tq.h"
 #include "sync.h"
+#include "tq.h"
+#include "tqCommon.h"
 #include "tsdb.h"
 #include "vnd.h"
-#include "tqCommon.h"
 
 #define BATCH_ENABLE 0
 
@@ -411,7 +411,7 @@ static int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
 }
 
 static int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) {
-  int32_t code = tmsgSendReq(pEpSet, pMsg);
+  int32_t code = tmsgSendSyncReq(pEpSet, pMsg);
   if (code != 0) {
     rpcFreeCont(pMsg->pCont);
     pMsg->pCont = NULL;
@@ -555,7 +555,7 @@ static void vnodeRestoreFinish(const SSyncFSM *pFsm, const SyncIndex commitIdx)
   walApplyVer(pVnode->pWal, commitIdx);
   pVnode->restored = true;
 
-  SStreamMeta* pMeta = pVnode->pTq->pStreamMeta;
+  SStreamMeta *pMeta = pVnode->pTq->pStreamMeta;
   streamMetaWLock(pMeta);
 
   if (pMeta->startInfo.tasksWillRestart) {

source/libs/transport/src/tmsgcb.c

@@ -48,6 +48,14 @@ int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) {
   }
   return code;
 }
+int32_t tmsgSendSyncReq(const SEpSet* epSet, SRpcMsg* pMsg) {
+  int32_t code = (*defaultMsgCb.sendSyncReqFp)(epSet, pMsg);
+  if (code != 0) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+  }
+  return code;
+}
 
 void tmsgSendRsp(SRpcMsg* pMsg) {
 #if 1

source/libs/transport/src/transCli.c

@@ -1907,7 +1907,12 @@ bool cliRecvReleaseReq(SCliConn* conn, STransMsgHead* pHead) {
 static void* cliWorkThread(void* arg) {
   SCliThrd* pThrd = (SCliThrd*)arg;
   pThrd->pid = taosGetSelfPthreadId();
-  setThreadName("trans-cli-work");
+
+  char    threadName[TSDB_LABEL_LEN] = {0};
+  STrans* pInst = pThrd->pTransInst;
+  strtolower(threadName, pInst->label);
+  setThreadName(threadName);
+
   uv_run(pThrd->loop, UV_RUN_DEFAULT);
 
   tDebug("thread quit-thread:%08" PRId64, pThrd->pid);
@@ -2701,6 +2706,7 @@ int transSendRecvWithTimeout(void* shandle, SEpSet* pEpSet, STransMsg* pReq, STr
     ret = TSDB_CODE_TIMEOUT_ERROR;
   } else {
     memcpy(pRsp, pSyncMsg->pRsp, sizeof(STransMsg));
+    pSyncMsg->pRsp->pCont = NULL;
     if (pSyncMsg->hasEpSet == 1) {
       epsetAssign(pEpSet, &pSyncMsg->epSet);
       *epUpdated = 1;

source/libs/transport/src/transComm.c

@@ -667,7 +667,7 @@ void transDestroySyncMsg(void* msg) {
   STransSyncMsg* pSyncMsg = msg;
   tsem_destroy(pSyncMsg->pSem);
   taosMemoryFree(pSyncMsg->pSem);
-
+  transFreeMsg(pSyncMsg->pRsp->pCont);
   taosMemoryFree(pSyncMsg->pRsp);
   taosMemoryFree(pSyncMsg);
 }