Merge pull request #18857 from taosdata/enh/changeDefaultValue

fix: change default sch value

source/libs/qworker/inc/qwInt.h

@@ -31,7 +31,7 @@ extern "C" {
 
 #define QW_DEFAULT_SCHEDULER_NUMBER 100
 #define QW_DEFAULT_TASK_NUMBER      10000
-#define QW_DEFAULT_SCH_TASK_NUMBER  10000
+#define QW_DEFAULT_SCH_TASK_NUMBER  3000
 #define QW_DEFAULT_SHORT_RUN_TIMES  2
 #define QW_DEFAULT_HEARTBEAT_MSEC   5000
 #define QW_SCH_TIMEOUT_MSEC         180000

source/libs/qworker/src/qwUtil.c

@@ -463,6 +463,8 @@ void qwDestroyImpl(void *pMgmt) {
   int8_t    nodeType = mgmt->nodeType;
   int32_t   nodeId = mgmt->nodeId;
 
+  int32_t taskCount = 0;
+  int32_t schStatusCount = 0;
   qDebug("start to destroy qworker, type:%d, id:%d, handle:%p", nodeType, nodeId, mgmt);
 
   taosTmrStop(mgmt->hbTimer);
@@ -472,6 +474,7 @@ void qwDestroyImpl(void *pMgmt) {
   uint64_t qId, tId;
   int32_t  eId;
   void    *pIter = taosHashIterate(mgmt->ctxHash, NULL);
+
   while (pIter) {
     SQWTaskCtx *ctx = (SQWTaskCtx *)pIter;
     void       *key = taosHashGetKey(pIter, NULL);
@@ -480,6 +483,7 @@ void qwDestroyImpl(void *pMgmt) {
     qwFreeTaskCtx(ctx);
     QW_TASK_DLOG_E("task ctx freed");
     pIter = taosHashIterate(mgmt->ctxHash, pIter);
+    taskCount++;
   }
   taosHashCleanup(mgmt->ctxHash);
 
@@ -487,7 +491,9 @@ void qwDestroyImpl(void *pMgmt) {
   while (pIter) {
     SQWSchStatus *sch = (SQWSchStatus *)pIter;
     qwDestroySchStatus(sch);
+
     pIter = taosHashIterate(mgmt->schHash, pIter);
+    schStatusCount++;
   }
   taosHashCleanup(mgmt->schHash);
 
@@ -499,7 +505,8 @@ void qwDestroyImpl(void *pMgmt) {
 
   qwCloseRef();
 
-  qDebug("qworker destroyed, type:%d, id:%d, handle:%p", nodeType, nodeId, mgmt);
+  qDebug("qworker destroyed, type:%d, id:%d, handle:%p, taskCount:%d, schStatusCount: %d", nodeType, nodeId, mgmt,
+         taskCount, schStatusCount);
 }
 
 int32_t qwOpenRef(void) {

source/libs/qworker/src/qworker.c

@@ -8,9 +8,9 @@
 #include "qwMsg.h"
 #include "tcommon.h"
 #include "tdatablock.h"
+#include "tglobal.h"
 #include "tmsg.h"
 #include "tname.h"
-#include "tglobal.h"
 
 SQWorkerMgmt gQwMgmt = {
     .lock = 0,
@@ -275,7 +275,7 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen,
           QW_ERR_RET(code);
         }
 
-        QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %"PRId64, pOutput->numOfBlocks,
+        QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %" PRId64, pOutput->numOfBlocks,
                      pOutput->numOfRows);
 
         qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC);
@@ -327,12 +327,14 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen,
     }
 
     if (0 == ctx->level) {
-      QW_TASK_DLOG("task fetched blocks %d rows %"PRId64", level %d", pOutput->numOfBlocks, pOutput->numOfRows, ctx->level);
+      QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 ", level %d", pOutput->numOfBlocks, pOutput->numOfRows,
+                   ctx->level);
       break;
     }
 
     if (pOutput->numOfRows >= QW_MIN_RES_ROWS) {
-      QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 " reaches the min rows", pOutput->numOfBlocks, pOutput->numOfRows);
+      QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 " reaches the min rows", pOutput->numOfBlocks,
+                   pOutput->numOfRows);
       break;
     }
   }
@@ -671,8 +673,8 @@ _return:
       qwBuildAndSendFetchRsp(ctx->fetchType, &qwMsg->connInfo, rsp, dataLen, code);
       rsp = NULL;
 
-      QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code,
-                   tstrerror(code), dataLen);
+      QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code),
+                   dataLen);
     }
   }
 
@@ -748,7 +750,8 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
     }
 
     QW_LOCK(QW_WRITE, &ctx->lock);
-    if ((queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code || 0 == atomic_load_8((int8_t *)&ctx->queryContinue)) {
+    if ((queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code ||
+        0 == atomic_load_8((int8_t *)&ctx->queryContinue)) {
       // Note: query is not running anymore
       QW_SET_PHASE(ctx, 0);
       QW_UNLOCK(QW_WRITE, &ctx->lock);