diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index c436e4ffd2ef8f68ea9a13484d5f167529614c9f..ddda8f8c9aafcdf3558525bd51666d9975f555c4 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -63,7 +63,7 @@ int32_t tsNumOfVnodeWriteThreads = 2; int32_t tsNumOfVnodeSyncThreads = 2; int32_t tsNumOfVnodeRsmaThreads = 2; int32_t tsNumOfQnodeQueryThreads = 4; -int32_t tsNumOfQnodeFetchThreads = 4; +int32_t tsNumOfQnodeFetchThreads = 1; int32_t tsNumOfSnodeSharedThreads = 2; int32_t tsNumOfSnodeUniqueThreads = 2; @@ -385,9 +385,9 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { tsNumOfQnodeQueryThreads = TMAX(tsNumOfQnodeQueryThreads, 4); if (cfgAddInt32(pCfg, "numOfQnodeQueryThreads", tsNumOfQnodeQueryThreads, 1, 1024, 0) != 0) return -1; - tsNumOfQnodeFetchThreads = tsNumOfCores / 2; - tsNumOfQnodeFetchThreads = TMAX(tsNumOfQnodeFetchThreads, 4); - if (cfgAddInt32(pCfg, "numOfQnodeFetchThreads", tsNumOfQnodeFetchThreads, 1, 1024, 0) != 0) return -1; +// tsNumOfQnodeFetchThreads = tsNumOfCores / 2; +// tsNumOfQnodeFetchThreads = TMAX(tsNumOfQnodeFetchThreads, 4); +// if (cfgAddInt32(pCfg, "numOfQnodeFetchThreads", tsNumOfQnodeFetchThreads, 1, 1024, 0) != 0) return -1; tsNumOfSnodeSharedThreads = tsNumOfCores / 4; tsNumOfSnodeSharedThreads = TRANGE(tsNumOfSnodeSharedThreads, 2, 4); @@ -527,6 +527,7 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) { pItem->stype = stype; } +/* pItem = cfgGetItem(tsCfg, "numOfQnodeFetchThreads"); if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { tsNumOfQnodeFetchThreads = numOfCores / 2; @@ -534,6 +535,7 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) { pItem->i32 = tsNumOfQnodeFetchThreads; pItem->stype = stype; } +*/ pItem = cfgGetItem(tsCfg, "numOfSnodeSharedThreads"); if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { @@ -691,7 +693,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32; tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; tsNumOfQnodeQueryThreads = cfgGetItem(pCfg, "numOfQnodeQueryThreads")->i32; - tsNumOfQnodeFetchThreads = cfgGetItem(pCfg, "numOfQnodeFetchThreads")->i32; +// tsNumOfQnodeFetchThreads = cfgGetItem(pCfg, "numOfQnodeFetchThreads")->i32; tsNumOfSnodeSharedThreads = cfgGetItem(pCfg, "numOfSnodeSharedThreads")->i32; tsNumOfSnodeUniqueThreads = cfgGetItem(pCfg, "numOfSnodeUniqueThreads")->i32; tsRpcQueueMemoryAllowed = cfgGetItem(pCfg, "rpcQueueMemoryAllowed")->i64; @@ -939,8 +941,10 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; } else if (strcasecmp("numOfQnodeQueryThreads", name) == 0) { tsNumOfQnodeQueryThreads = cfgGetItem(pCfg, "numOfQnodeQueryThreads")->i32; +/* } else if (strcasecmp("numOfQnodeFetchThreads", name) == 0) { tsNumOfQnodeFetchThreads = cfgGetItem(pCfg, "numOfQnodeFetchThreads")->i32; +*/ } else if (strcasecmp("numOfSnodeSharedThreads", name) == 0) { tsNumOfSnodeSharedThreads = cfgGetItem(pCfg, "numOfSnodeSharedThreads")->i32; } else if (strcasecmp("numOfSnodeUniqueThreads", name) == 0) { diff --git a/source/libs/scheduler/inc/schInt.h b/source/libs/scheduler/inc/schInt.h index 957fd46ba5a767858a3bb5bbe50142b4f1c1ce47..7fea2867323f9d5f119740d94a80da4f1a0b5b51 100644 --- a/source/libs/scheduler/inc/schInt.h +++ b/source/libs/scheduler/inc/schInt.h @@ -283,8 +283,9 @@ typedef struct SSchJob { } SSchJob; typedef struct SSchTaskCtx { - int64_t jobRid; + int64_t jobRid; SSchTask *pTask; + bool asyncLaunch; } SSchTaskCtx; extern SSchedulerMgmt schMgmt; diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index 5a64aaaebb3860d2c6729ac8eb1e00be0cc9cda1..275fbb8016b79a457cc3621632795fd200b1dec2 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -396,7 +396,7 @@ int32_t schHandleCallback(void *param, SDataBuf *pMsg, int32_t rspCode) { tstrerror(rspCode)); SCH_ERR_JRET(schProcessOnCbBegin(&pJob, &pTask, pParam->queryId, pParam->refId, pParam->taskId)); - + code = schHandleResponseMsg(pJob, pTask, pParam->execId, pMsg, rspCode); pMsg->pData = NULL; diff --git a/source/libs/scheduler/src/schTask.c b/source/libs/scheduler/src/schTask.c index c061eb078981a63c19eec563863b6aad77f50fd6..8b3a2016b80bb1c0f96bc312f2043f370f5b7c54 100644 --- a/source/libs/scheduler/src/schTask.c +++ b/source/libs/scheduler/src/schTask.c @@ -138,12 +138,6 @@ int32_t schUpdateTaskExecNode(SSchJob *pJob, SSchTask *pTask, void *handle, int3 return TSDB_CODE_SUCCESS; } - if ((execId != pTask->execId) || pTask->waitRetry) { // ignore it - SCH_TASK_DLOG("handle not updated since execId %d is already not current execId %d, waitRetry %d", execId, - pTask->execId, pTask->waitRetry); - return TSDB_CODE_SUCCESS; - } - SSchNodeInfo *nodeInfo = taosHashGet(pTask->execNodes, &execId, sizeof(execId)); if (NULL == nodeInfo) { // ignore it SCH_TASK_DLOG("handle not updated since execId %d already not exist, current execId %d, waitRetry %d", execId, @@ -162,11 +156,16 @@ int32_t schUpdateTaskHandle(SSchJob *pJob, SSchTask *pTask, bool dropExecNode, v if (dropExecNode) { SCH_RET(schDropTaskExecNode(pJob, pTask, handle, execId)); } + + schUpdateTaskExecNode(pJob, pTask, handle, execId); + if ((execId != pTask->execId) || pTask->waitRetry) { // ignore it + SCH_TASK_DLOG("handle not updated since execId %d is already not current execId %d, waitRetry %d", execId, pTask->execId, pTask->waitRetry); + SCH_ERR_RET(TSDB_CODE_SCH_IGNORE_ERROR); + } + SCH_SET_TASK_HANDLE(pTask, handle); - schUpdateTaskExecNode(pJob, pTask, handle, execId); - return TSDB_CODE_SUCCESS; } @@ -352,7 +351,7 @@ int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32 pTask->waitRetry = true; schDropTaskOnExecNode(pJob, pTask); taosHashClear(pTask->execNodes); - SCH_ERR_JRET(schRemoveTaskFromExecList(pJob, pTask)); + schRemoveTaskFromExecList(pJob, pTask); schDeregisterTaskHb(pJob, pTask); atomic_sub_fetch_32(&pTask->level->taskLaunchedNum, 1); taosMemoryFreeClear(pTask->msg); @@ -599,7 +598,7 @@ int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bo int32_t schHandleTaskRetry(SSchJob *pJob, SSchTask *pTask) { atomic_sub_fetch_32(&pTask->level->taskLaunchedNum, 1); - SCH_ERR_RET(schRemoveTaskFromExecList(pJob, pTask)); + schRemoveTaskFromExecList(pJob, pTask); SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_INIT); if (SCH_TASK_NEED_FLOW_CTRL(pJob, pTask)) { @@ -746,8 +745,7 @@ _return: int32_t schRemoveTaskFromExecList(SSchJob *pJob, SSchTask *pTask) { int32_t code = taosHashRemove(pJob->execTasks, &pTask->taskId, sizeof(pTask->taskId)); if (code) { - SCH_TASK_ELOG("task failed to rm from execTask list, code:%x", code); - SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR); + SCH_TASK_WLOG("task already not in execTask list, code:%x", code); } return TSDB_CODE_SUCCESS; @@ -836,6 +834,11 @@ int32_t schLaunchTaskImpl(void *param) { } SSchTask *pTask = pCtx->pTask; + + if (pCtx->asyncLaunch) { + SCH_LOCK_TASK(pTask); + } + int8_t status = 0; int32_t code = 0; @@ -882,8 +885,6 @@ int32_t schLaunchTaskImpl(void *param) { _return: - taosMemoryFree(param); - if (pJob->taskNum >= SCH_MIN_AYSNC_EXEC_NUM) { if (code) { code = schProcessOnTaskFailure(pJob, pTask, code); @@ -893,8 +894,14 @@ _return: } } + if (pCtx->asyncLaunch) { + SCH_UNLOCK_TASK(pTask); + } + schReleaseJob(pJob->refId); + taosMemoryFree(param); + SCH_RET(code); } @@ -908,6 +915,7 @@ int32_t schAsyncLaunchTaskImpl(SSchJob *pJob, SSchTask *pTask) { param->pTask = pTask; if (pJob->taskNum >= SCH_MIN_AYSNC_EXEC_NUM) { + param->asyncLaunch = true; taosAsyncExec(schLaunchTaskImpl, param, NULL); } else { SCH_ERR_RET(schLaunchTaskImpl(param));