Merge branch 'main' into feature/3_liaohj

include/common/tcommon.h

@@ -300,6 +300,7 @@ typedef struct STableBlockDistInfo {
   int32_t  firstSeekTimeUs;
   uint32_t numOfInmemRows;
   uint32_t numOfSmallBlocks;
+  uint32_t numOfVgroups;
   int32_t  blockRowsHisto[20];
 } STableBlockDistInfo;
 

include/util/tworker.h

@@ -26,12 +26,12 @@ extern "C" {
 typedef struct SQWorkerPool SQWorkerPool;
 typedef struct SWWorkerPool SWWorkerPool;
 
-typedef struct SQWorker {
+typedef struct SQueueWorker {
   int32_t  id;      // worker id
   int64_t  pid;     // thread pid
   TdThread thread;  // thread id
   void    *pool;
-} SQWorker;
+} SQueueWorker;
 
 typedef struct SQWorkerPool {
   int32_t       max;  // max number of workers
@@ -39,7 +39,7 @@ typedef struct SQWorkerPool {
   int32_t       num;  // current number of workers
   STaosQset    *qset;
   const char   *name;
-  SQWorker     *workers;
+  SQueueWorker *workers;
   TdThreadMutex mutex;
 } SQWorkerPool;
 

source/client/src/clientHb.c

@@ -347,7 +347,7 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
       continue;
     }
 
-    if (pRequest->killed) {
+    if (pRequest->killed || 0 == pRequest->body.queryJob) {
       releaseRequest(*rid);
       pIter = taosHashIterate(pObj->pRequests, pIter);
       continue;

source/dnode/mnode/impl/inc/mndInt.h

@@ -58,7 +58,7 @@ typedef int32_t (*MndInitFp)(SMnode *pMnode);
 typedef void (*MndCleanupFp)(SMnode *pMnode);
 typedef int32_t (*ShowRetrieveFp)(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter);
-typedef struct SQWorker SQHandle;
+typedef struct SQueueWorker SQHandle;
 
 typedef struct {
   const char  *name;

source/dnode/qnode/inc/qndInt.h

@@ -29,7 +29,7 @@
 extern "C" {
 #endif
 
-typedef struct SQWorker SQHandle;
+typedef struct SQueueWorker SQHandle;
 
 typedef struct SQnode {
   int32_t   qndId;

source/dnode/vnode/src/inc/vnodeInt.h

@@ -58,7 +58,7 @@ typedef struct STQ                STQ;
 typedef struct SVState            SVState;
 typedef struct SVStatis           SVStatis;
 typedef struct SVBufPool          SVBufPool;
-typedef struct SQWorker           SQHandle;
+typedef struct SQueueWorker       SQHandle;
 typedef struct STsdbKeepCfg       STsdbKeepCfg;
 typedef struct SMetaSnapReader    SMetaSnapReader;
 typedef struct SMetaSnapWriter    SMetaSnapWriter;

source/libs/executor/src/sysscanoperator.c

@@ -350,8 +350,8 @@ static int32_t optSysMergeRslt(SArray* mRslt, SArray* rslt);
 static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableScanInfo* pInfo, const char* name,
                                           SExecTaskInfo* pTaskInfo);
 void                extractTbnameSlotId(SSysTableScanInfo* pInfo, const SScanPhysiNode* pScanNode);
-static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo,
-                                                 const char* name, SSDataBlock* pBlock);
+static void         sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, const char* name,
+                                           SSDataBlock* pBlock);
 __optSysFilter      optSysGetFilterFunc(int32_t ctype, bool* reverse) {
   if (ctype == OP_TYPE_LOWER_EQUAL || ctype == OP_TYPE_LOWER_THAN) {
     *reverse = true;
@@ -516,7 +516,8 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
       metaTbCursorPrev(pInfo->pCur);
       blockFull = true;
     } else {
-      sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows, dataBlock);
+      sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows,
+                                       dataBlock);
     }
 
     metaReaderClear(&smrSuperTable);
@@ -1343,7 +1344,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
       setOperatorCompleted(pOperator);
     }
 
-    return pBlock->info.rows > 0? pBlock:NULL;
+    return pBlock->info.rows > 0 ? pBlock : NULL;
   } else {
     return NULL;
   }
@@ -1489,10 +1490,11 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
   setOperatorInfo(pOperator, "SysTableScanOperator", QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, false, OP_NOT_OPENED,
                   pInfo, pTaskInfo);
   pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
-  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, optrDefaultBufFn, NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(optrDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, optrDefaultBufFn, NULL);
   return pOperator;
 
-  _error:
+_error:
   if (pInfo != NULL) {
     destroySysScanOperator(pInfo);
   }
@@ -2001,7 +2003,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
   }
 
   pInfo->readHandle = *readHandle;
-  pInfo->uid = (pBlockScanNode->suid != 0)? pBlockScanNode->suid:pBlockScanNode->uid;
+  pInfo->uid = (pBlockScanNode->suid != 0) ? pBlockScanNode->suid : pBlockScanNode->uid;
 
   int32_t    numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols);
@@ -2012,8 +2014,8 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
 
   setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false,
                   OP_NOT_OPENED, pInfo, pTaskInfo);
-  pOperator->fpSet =
-      createOperatorFpSet(optrDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, optrDefaultBufFn, NULL);
+  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo,
+                                         optrDefaultBufFn, NULL);
   return pOperator;
 
 _error:

source/libs/function/src/builtinsimpl.c

@@ -795,7 +795,7 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
       case TSDB_DATA_TYPE_UBIGINT:
       case TSDB_DATA_TYPE_BIGINT:
         ((int64_t*)pCol->pData)[currentRow] = pRes->v;
-//        colDataAppendInt64(pCol, currentRow, &pRes->v);
+        //        colDataAppendInt64(pCol, currentRow, &pRes->v);
         break;
       case TSDB_DATA_TYPE_UINT:
       case TSDB_DATA_TYPE_INT:
@@ -2123,7 +2123,8 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
   }
 
   // All null data column, return directly.
-  if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && pInputCol->hasNull == true) {
+  if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) &&
+      pInputCol->hasNull == true) {
     // save selectivity value for column consisted of all null values
     int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
     if (code != TSDB_CODE_SUCCESS) {
@@ -2239,7 +2240,8 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
   }
 
   // All null data column, return directly.
-  if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && pInputCol->hasNull == true) {
+  if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) &&
+      pInputCol->hasNull == true) {
     // save selectivity value for column consisted of all null values
     int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
     if (code != TSDB_CODE_SUCCESS) {
@@ -2875,8 +2877,8 @@ static STopBotRes* getTopBotOutputInfo(SqlFunctionCtx* pCtx) {
   return pRes;
 }
 
-static int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
-                               uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery);
+static int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock,
+                               uint16_t type, uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery);
 
 static void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type, bool isTopQuery);
 
@@ -5410,6 +5412,7 @@ int32_t blockDistFunction(SqlFunctionCtx* pCtx) {
   if (pDistInfo->maxRows < p1.maxRows) {
     pDistInfo->maxRows = p1.maxRows;
   }
+  pDistInfo->numOfVgroups += (p1.numOfTables != 0 ? 1 : 0);
 
   for (int32_t i = 0; i < tListLen(pDistInfo->blockRowsHisto); ++i) {
     pDistInfo->blockRowsHisto[i] += p1.blockRowsHisto[i];
@@ -5438,6 +5441,7 @@ int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDist
   if (tEncodeI32(&encoder, pInfo->defMinRows) < 0) return -1;
   if (tEncodeU32(&encoder, pInfo->numOfInmemRows) < 0) return -1;
   if (tEncodeU32(&encoder, pInfo->numOfSmallBlocks) < 0) return -1;
+  if (tEncodeU32(&encoder, pInfo->numOfVgroups) < 0) return -1;
 
   for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) {
     if (tEncodeI32(&encoder, pInfo->blockRowsHisto[i]) < 0) return -1;
@@ -5469,6 +5473,7 @@ int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo
   if (tDecodeI32(&decoder, &pInfo->defMinRows) < 0) return -1;
   if (tDecodeU32(&decoder, &pInfo->numOfInmemRows) < 0) return -1;
   if (tDecodeU32(&decoder, &pInfo->numOfSmallBlocks) < 0) return -1;
+  if (tDecodeU32(&decoder, &pInfo->numOfVgroups) < 0) return -1;
 
   for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) {
     if (tDecodeI32(&decoder, &pInfo->blockRowsHisto[i]) < 0) return -1;
@@ -5520,7 +5525,7 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   colDataAppend(pColInfo, row++, st, false);
 
   len = sprintf(st + VARSTR_HEADER_SIZE, "Total_Tables=[%d] Total_Files=[%d] Total_Vgroups=[%d]", pData->numOfTables,
-                pData->numOfFiles, 0);
+                pData->numOfFiles, pData->numOfVgroups);
 
   varDataSetLen(st, len);
   colDataAppend(pColInfo, row++, st, false);

source/libs/qworker/inc/qwInt.h

@@ -76,6 +76,7 @@ typedef struct SQWDebug {
   bool lockEnable;
   bool statusEnable;
   bool dumpEnable;
+  bool forceStop;
   bool sleepSimulate;
   bool deadSimulate;
   bool redirectSimulate;
@@ -248,6 +249,7 @@ typedef struct SQWorkerMgmt {
 
 #define QW_QUERY_RUNNING(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY)
 #define QW_FETCH_RUNNING(ctx) ((ctx)->inFetch)
+#define QW_QUERY_NOT_STARTED(ctx) (QW_GET_PHASE(ctx) == -1)
 
 #define QW_SET_QTID(id, qId, tId, eId)                              \
   do {                                                              \

source/libs/qworker/src/qwDbg.c

@@ -9,11 +9,13 @@
 #include "tmsg.h"
 #include "tname.h"
 
-SQWDebug gQWDebug = {.statusEnable = true,
+SQWDebug gQWDebug = {.lockEnable = false,
+                     .statusEnable = true,
                      .dumpEnable = false,
                      .redirectSimulate = false,
                      .deadSimulate = false,
-                     .sleepSimulate = false};
+                     .sleepSimulate = false,
+                     .forceStop = false};
 
 int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore) {
   if (!gQWDebug.statusEnable) {
@@ -306,6 +308,12 @@ int32_t qwDbgEnableDebug(char *option) {
     return TSDB_CODE_SUCCESS;
   }
 
+  if (0 == strcasecmp(option, "forceStop")) {
+    gQWDebug.forceStop = true;
+    qError("qw forceStop debug enabled");
+    return TSDB_CODE_SUCCESS;
+  }
+
   qError("invalid qw debug option:%s", option);
 
   return TSDB_CODE_APP_ERROR;

source/libs/qworker/src/qworker.c

@@ -18,6 +18,51 @@ SQWorkerMgmt gQwMgmt = {
     .qwNum = 0,
 };
 
+
+int32_t qwStopAllTasks(SQWorker *mgmt) {
+  uint64_t qId, tId, sId;
+  int32_t  eId;
+  int64_t  rId = 0;
+
+  void    *pIter = taosHashIterate(mgmt->ctxHash, NULL);
+  while (pIter) {
+    SQWTaskCtx *ctx = (SQWTaskCtx *)pIter;
+    void       *key = taosHashGetKey(pIter, NULL);
+    QW_GET_QTID(key, qId, tId, eId);
+
+    QW_LOCK(QW_WRITE, &ctx->lock);
+
+    sId = ctx->sId;
+
+    QW_TASK_DLOG_E("start to force stop task");
+
+    if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP) || QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
+      QW_TASK_WLOG_E("task already dropping");
+      QW_UNLOCK(QW_WRITE, &ctx->lock);
+
+      pIter = taosHashIterate(mgmt->ctxHash, pIter);
+      continue;
+    }
+
+    if (QW_QUERY_RUNNING(ctx)) {
+      qwKillTaskHandle(ctx, TSDB_CODE_VND_STOPPED);
+      QW_TASK_DLOG_E("task running, async killed");
+    } else if (QW_FETCH_RUNNING(ctx)) {
+      QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED);
+      QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);    
+      QW_TASK_DLOG_E("task fetching, update drop received");
+    } else {
+      qwDropTask(QW_FPARAMS());
+    }
+
+    QW_UNLOCK(QW_WRITE, &ctx->lock);
+
+    pIter = taosHashIterate(mgmt->ctxHash, pIter);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
   int32_t         code = 0;
   SSchedulerHbRsp rsp = {0};
@@ -973,6 +1018,10 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
 
   qwDbgDumpMgmtInfo(mgmt);
 
+  if (gQWDebug.forceStop) {
+    (void)qwStopAllTasks(mgmt);
+  }
+
   QW_LOCK(QW_READ, &mgmt->schLock);
 
   int32_t schNum = taosHashGetSize(mgmt->schHash);
@@ -1087,6 +1136,7 @@ _return:
   QW_RET(TSDB_CODE_SUCCESS);
 }
 
+
 int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, void **qWorkerMgmt, const SMsgCb *pMsgCb) {
   if (NULL == qWorkerMgmt || (pMsgCb && pMsgCb->mgmt == NULL)) {
     qError("invalid param to init qworker");
@@ -1186,45 +1236,9 @@ void qWorkerStopAllTasks(void *qWorkerMgmt) {
 
   QW_DLOG("start to stop all tasks, taskNum:%d", taosHashGetSize(mgmt->ctxHash));
  
-  uint64_t qId, tId, sId;
-  int32_t  eId;
-  int64_t  rId = 0;
-  
   atomic_store_8(&mgmt->nodeStopped, 1);
 
-  void    *pIter = taosHashIterate(mgmt->ctxHash, NULL);
-  while (pIter) {
-    SQWTaskCtx *ctx = (SQWTaskCtx *)pIter;
-    void       *key = taosHashGetKey(pIter, NULL);
-    QW_GET_QTID(key, qId, tId, eId);
-
-    QW_LOCK(QW_WRITE, &ctx->lock);
-
-    sId = ctx->sId;
-
-    QW_TASK_DLOG_E("start to force stop task");
-
-    if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP) || QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
-      QW_TASK_WLOG_E("task already dropping");
-      QW_UNLOCK(QW_WRITE, &ctx->lock);
-
-      pIter = taosHashIterate(mgmt->ctxHash, pIter);
-      continue;
-    }
-
-    if (QW_QUERY_RUNNING(ctx)) {
-      qwKillTaskHandle(ctx, TSDB_CODE_VND_STOPPED);
-    } else if (QW_FETCH_RUNNING(ctx)) {
-      QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED);
-      QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);    
-    } else {
-      qwDropTask(QW_FPARAMS());
-    }
-
-    QW_UNLOCK(QW_WRITE, &ctx->lock);
-
-    pIter = taosHashIterate(mgmt->ctxHash, pIter);
-  }
+  (void)qwStopAllTasks(mgmt);
 }
 
 void qWorkerDestroy(void **qWorkerMgmt) {

source/libs/transport/src/transCli.c

@@ -727,7 +727,7 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
   QUEUE_INIT(&conn->q);
   conn->hostThrd = pThrd;
   conn->status = ConnNormal;
-  conn->broken = 0;
+  conn->broken = false;
   transRefCliHandle(conn);
 
   atomic_add_fetch_32(&pThrd->connCount, 1);
@@ -997,6 +997,11 @@ static void cliDestroyBatch(SCliBatch* pBatch) {
   taosMemoryFree(pBatch);
 }
 static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
+  if (pThrd->quit == true) {
+    cliDestroyBatch(pBatch);
+    return;
+  }
+
   if (pBatch == NULL || pBatch->wLen == 0 || QUEUE_IS_EMPTY(&pBatch->wq)) {
     return;
   }
@@ -1082,18 +1087,24 @@ static void cliSendBatchCb(uv_write_t* req, int status) {
   if (status != 0) {
     tDebug("%s conn %p failed to send batch msg, batch size:%d, msgLen:%d, reason:%s", CONN_GET_INST_LABEL(conn), conn,
            p->wLen, p->batchSize, uv_err_name(status));
-    cliHandleExcept(conn);
+
+    if (!uv_is_closing((uv_handle_t*)&conn->stream)) cliHandleExcept(conn);
+
     cliHandleBatchReq(nxtBatch, thrd);
   } else {
     tDebug("%s conn %p succ to send batch msg, batch size:%d, msgLen:%d", CONN_GET_INST_LABEL(conn), conn, p->wLen,
            p->batchSize);
-
+    if (!uv_is_closing((uv_handle_t*)&conn->stream)) {
       if (nxtBatch != NULL) {
         conn->pBatch = nxtBatch;
         cliSendBatch(conn);
       } else {
         addConnToPool(thrd->pool, conn);
       }
+    } else {
+      cliDestroyBatch(nxtBatch);
+      // conn release by other callback
+    }
   }
 
   cliDestroyBatch(p);
@@ -1454,6 +1465,11 @@ static void cliNoBatchDealReq(queue* wq, SCliThrd* pThrd) {
     QUEUE_REMOVE(h);
 
     SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
+
+    if (pMsg->type == Quit) {
+      pThrd->stopMsg = pMsg;
+      continue;
+    }
     (*cliAsyncHandle[pMsg->type])(pMsg, pThrd);
 
     count++;
@@ -1485,6 +1501,12 @@ static void cliBatchDealReq(queue* wq, SCliThrd* pThrd) {
     QUEUE_REMOVE(h);
 
     SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
+
+    if (pMsg->type == Quit) {
+      pThrd->stopMsg = pMsg;
+      continue;
+    }
+
     if (pMsg->type == Normal && REQUEST_NO_RESP(&pMsg->msg)) {
       STransConnCtx* pCtx = pMsg->ctx;
 
@@ -1582,7 +1604,6 @@ static void cliAsyncCb(uv_async_t* handle) {
   SCliThrd*   pThrd = item->pThrd;
   STrans*     pTransInst = pThrd->pTransInst;
 
-  SCliMsg* pMsg = NULL;
   // batch process to avoid to lock/unlock frequently
   queue wq;
   taosThreadMutexLock(&item->mtx);
@@ -2285,22 +2306,8 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran
     transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
     return TSDB_CODE_RPC_BROKEN_LINK;
   }
-  /*if (pTransInst->connLimitNum > 0 && REQUEST_NO_RESP(pReq)) {
-    char     key[TSDB_FQDN_LEN + 64] = {0};
-    char*    ip = EPSET_GET_INUSE_IP((SEpSet*)pEpSet);
-    uint16_t port = EPSET_GET_INUSE_PORT((SEpSet*)pEpSet);
-    CONN_CONSTRUCT_HASH_KEY(key, ip, port);
-
-    int32_t* val = taosHashGet(pThrd->connLimitCache, key, strlen(key));
-    if (val != NULL && *val >= pTransInst->connLimitNum) {
-      transFreeMsg(pReq->pCont);
-      transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-      return TSDB_CODE_RPC_MAX_SESSIONS;
-    }
-  }*/
 
   TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
-
   STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
   pCtx->epSet = *pEpSet;
   pCtx->origEpSet = *pEpSet;

source/util/src/tworker.c

@@ -22,7 +22,7 @@ typedef void *(*ThreadFp)(void *param);
 
 int32_t tQWorkerInit(SQWorkerPool *pool) {
   pool->qset = taosOpenQset();
-  pool->workers = taosMemoryCalloc(pool->max, sizeof(SQWorker));
+  pool->workers = taosMemoryCalloc(pool->max, sizeof(SQueueWorker));
   if (pool->workers == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -31,7 +31,7 @@ int32_t tQWorkerInit(SQWorkerPool *pool) {
   (void)taosThreadMutexInit(&pool->mutex, NULL);
 
   for (int32_t i = 0; i < pool->max; ++i) {
-    SQWorker *worker = pool->workers + i;
+    SQueueWorker *worker = pool->workers + i;
     worker->id = i;
     worker->pool = pool;
   }
@@ -42,14 +42,14 @@ int32_t tQWorkerInit(SQWorkerPool *pool) {
 
 void tQWorkerCleanup(SQWorkerPool *pool) {
   for (int32_t i = 0; i < pool->max; ++i) {
-    SQWorker *worker = pool->workers + i;
+    SQueueWorker *worker = pool->workers + i;
     if (taosCheckPthreadValid(worker->thread)) {
       taosQsetThreadResume(pool->qset);
     }
   }
 
   for (int32_t i = 0; i < pool->max; ++i) {
-    SQWorker *worker = pool->workers + i;
+    SQueueWorker *worker = pool->workers + i;
     if (taosCheckPthreadValid(worker->thread)) {
       uInfo("worker:%s:%d is stopping", pool->name, worker->id);
       taosThreadJoin(worker->thread, NULL);
@@ -65,7 +65,7 @@ void tQWorkerCleanup(SQWorkerPool *pool) {
   uInfo("worker:%s is closed", pool->name);
 }
 
-static void *tQWorkerThreadFp(SQWorker *worker) {
+static void *tQWorkerThreadFp(SQueueWorker *worker) {
   SQWorkerPool *pool = worker->pool;
   SQueueInfo    qinfo = {0};
   void         *msg = NULL;
@@ -106,7 +106,7 @@ STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
   // spawn a thread to process queue
   if (pool->num < pool->max) {
     do {
-      SQWorker *worker = pool->workers + pool->num;
+      SQueueWorker *worker = pool->workers + pool->num;
 
       TdThreadAttr thAttr;
       taosThreadAttrInit(&thAttr);
@@ -138,7 +138,7 @@ void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) {
 
 int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) {
   pool->qset = taosOpenQset();
-  pool->workers = taosArrayInit(2, sizeof(SQWorker *));
+  pool->workers = taosArrayInit(2, sizeof(SQueueWorker *));
   if (pool->workers == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -153,14 +153,14 @@ int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) {
 void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) {
   int32_t size = taosArrayGetSize(pool->workers);
   for (int32_t i = 0; i < size; ++i) {
-    SQWorker *worker = taosArrayGetP(pool->workers, i);
+    SQueueWorker *worker = taosArrayGetP(pool->workers, i);
     if (taosCheckPthreadValid(worker->thread)) {
       taosQsetThreadResume(pool->qset);
     }
   }
 
   for (int32_t i = 0; i < size; ++i) {
-    SQWorker *worker = taosArrayGetP(pool->workers, i);
+    SQueueWorker *worker = taosArrayGetP(pool->workers, i);
     if (taosCheckPthreadValid(worker->thread)) {
       uInfo("worker:%s:%d is stopping", pool->name, worker->id);
       taosThreadJoin(worker->thread, NULL);
@@ -177,7 +177,7 @@ void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) {
   uInfo("worker:%s is closed", pool->name);
 }
 
-static void *tAutoQWorkerThreadFp(SQWorker *worker) {
+static void *tAutoQWorkerThreadFp(SQueueWorker *worker) {
   SAutoQWorkerPool *pool = worker->pool;
   SQueueInfo        qinfo = {0};
   void             *msg = NULL;
@@ -222,7 +222,7 @@ STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem
 
   // spawn a thread to process queue
   while (curWorkerNum < dstWorkerNum) {
-    SQWorker *worker = taosMemoryCalloc(1, sizeof(SQWorker));
+    SQueueWorker *worker = taosMemoryCalloc(1, sizeof(SQueueWorker));
     if (worker == NULL || taosArrayPush(pool->workers, &worker) == NULL) {
       uError("worker:%s:%d failed to create", pool->name, curWorkerNum);
       taosMemoryFree(worker);