From 5f010843a654d04d9420d496a86ec7a0170dd8f0 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Thu, 13 Oct 2022 16:58:43 +0800 Subject: [PATCH 1/2] more code format --- source/dnode/vnode/src/tsdb/tsdbMergeTree.c | 20 +- source/dnode/vnode/src/tsdb/tsdbRead.c | 2 +- source/libs/scheduler/inc/schInt.h | 627 ++++++++++-------- source/libs/scheduler/src/schDbg.c | 6 +- source/libs/scheduler/src/schFlowCtrl.c | 121 ++-- source/libs/scheduler/src/schJob.c | 151 ++--- source/libs/scheduler/src/schStatus.c | 16 +- source/libs/scheduler/src/schTask.c | 63 +- source/libs/scheduler/src/schUtil.c | 53 +- source/libs/scheduler/src/scheduler.c | 12 +- source/libs/scheduler/test/schedulerTests.cpp | 382 +++++------ tools/scripts/codeFormat.sh | 22 +- 12 files changed, 732 insertions(+), 743 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbMergeTree.c b/source/dnode/vnode/src/tsdb/tsdbMergeTree.c index 4d54fbbd79..0b7d1cbaf4 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMergeTree.c +++ b/source/dnode/vnode/src/tsdb/tsdbMergeTree.c @@ -101,8 +101,8 @@ void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) { static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) { int32_t code = 0; - SSttBlockLoadInfo* pInfo = pIter->pBlockLoadInfo; - if (pInfo->blockIndex[0] == pIter->iSttBlk) { + SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo; + if (pInfo->blockIndex[0] == pIter->iSttBlk) { if (pInfo->currentLoadBlockIndex != 0) { tsdbDebug("current load index is set to 0, block index:%d, file index:%d, due to uid:%" PRIu64 ", load data, %s", pIter->iSttBlk, pIter->iStt, pIter->uid, idStr); @@ -113,7 +113,7 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) { if (pInfo->blockIndex[1] == pIter->iSttBlk) { if (pInfo->currentLoadBlockIndex != 1) { - tsdbDebug("current load index is set to 1, block index:%d, file index:%d, due to uid:%"PRIu64", load data, %s", + tsdbDebug("current load index is set to 1, block index:%d, file index:%d, due to uid:%" PRIu64 ", load data, %s", pIter->iSttBlk, pIter->iStt, pIter->uid, idStr); pInfo->currentLoadBlockIndex = 1; } @@ -140,8 +140,10 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) { pInfo->elapsedTime += el; pInfo->loadBlocks += 1; - tsdbDebug("read last block, total load:%d, trigger by uid:%"PRIu64", last file index:%d, last block index:%d, entry:%d, %p, elapsed time:%.2f ms, %s", - pInfo->loadBlocks, pIter->uid, pIter->iStt, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock, el, idStr); + tsdbDebug("read last block, total load:%d, trigger by uid:%" PRIu64 + ", last file index:%d, last block index:%d, entry:%d, %p, elapsed time:%.2f ms, %s", + pInfo->loadBlocks, pIter->uid, pIter->iStt, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock, el, + idStr); if (code != TSDB_CODE_SUCCESS) { goto _exit; } @@ -336,7 +338,7 @@ _exit: void tLDataIterClose(SLDataIter *pIter) { taosMemoryFree(pIter); } -void tLDataIterNextBlock(SLDataIter *pIter, const char* idStr) { +void tLDataIterNextBlock(SLDataIter *pIter, const char *idStr) { int32_t step = pIter->backward ? -1 : 1; int32_t oldIndex = pIter->iSttBlk; @@ -386,10 +388,10 @@ void tLDataIterNextBlock(SLDataIter *pIter, const char* idStr) { if (index != -1) { pIter->iSttBlk = index; pIter->pSttBlk = (SSttBlk *)taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, pIter->iSttBlk); - tsdbDebug("try next last file block:%d from %d, trigger by uid:%"PRIu64", file index:%d, %s", pIter->iSttBlk, oldIndex, pIter->uid, pIter->iStt, - idStr); + tsdbDebug("try next last file block:%d from %d, trigger by uid:%" PRIu64 ", file index:%d, %s", pIter->iSttBlk, + oldIndex, pIter->uid, pIter->iStt, idStr); } else { - tsdbDebug("no more last block qualified, uid:%"PRIu64", file index::%d, %s", pIter->uid, oldIndex, idStr); + tsdbDebug("no more last block qualified, uid:%" PRIu64 ", file index::%d, %s", pIter->uid, oldIndex, idStr); } } diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 4bb1f2d2fe..a7850f3114 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -1645,7 +1645,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, STSRow* pTSRow = NULL; SRowMerger merge = {0}; TSDBROW fRow = tMergeTreeGetRow(&pLastBlockReader->mergeTree); - tsdbTrace("fRow ptr:%p, %d, uid:%"PRIu64", %s", fRow.pBlockData, fRow.iRow, pLastBlockReader->uid, pReader->idStr); + tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", fRow.pBlockData, fRow.iRow, pLastBlockReader->uid, pReader->idStr); // only last block exists if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) { diff --git a/source/libs/scheduler/inc/schInt.h b/source/libs/scheduler/inc/schInt.h index 7ced4f626c..1a88992840 100644 --- a/source/libs/scheduler/inc/schInt.h +++ b/source/libs/scheduler/inc/schInt.h @@ -20,13 +20,13 @@ extern "C" { #endif +#include "command.h" #include "os.h" -#include "tarray.h" #include "planner.h" #include "scheduler.h" +#include "tarray.h" #include "thash.h" #include "trpc.h" -#include "command.h" enum { SCH_READ = 1, @@ -51,20 +51,20 @@ typedef enum { SCH_ALL, } SCH_POLICY; -#define SCHEDULE_DEFAULT_MAX_JOB_NUM 1000 -#define SCHEDULE_DEFAULT_MAX_TASK_NUM 1000 +#define SCHEDULE_DEFAULT_MAX_JOB_NUM 1000 +#define SCHEDULE_DEFAULT_MAX_TASK_NUM 1000 #define SCHEDULE_DEFAULT_MAX_NODE_TABLE_NUM 200 // unit is TSDB_TABLE_NUM_UNIT -#define SCHEDULE_DEFAULT_POLICY SCH_LOAD_SEQ -#define SCHEDULE_DEFAULT_MAX_NODE_NUM 20 +#define SCHEDULE_DEFAULT_POLICY SCH_LOAD_SEQ +#define SCHEDULE_DEFAULT_MAX_NODE_NUM 20 #define SCH_DEFAULT_TASK_TIMEOUT_USEC 10000000 -#define SCH_MAX_TASK_TIMEOUT_USEC 60000000 -#define SCH_DEFAULT_MAX_RETRY_NUM 6 -#define SCH_MIN_AYSNC_EXEC_NUM 3 +#define SCH_MAX_TASK_TIMEOUT_USEC 60000000 +#define SCH_DEFAULT_MAX_RETRY_NUM 6 +#define SCH_MIN_AYSNC_EXEC_NUM 3 typedef struct SSchDebug { - bool lockEnable; - bool apiEnable; + bool lockEnable; + bool apiEnable; } SSchDebug; typedef struct SSchTrans { @@ -80,7 +80,6 @@ typedef struct SSchHbTrans { } SSchHbTrans; typedef struct SSchApiStat { - #if defined(WINDOWS) || defined(_TD_DARWIN_64) size_t avoidCompilationErrors; #endif @@ -88,7 +87,6 @@ typedef struct SSchApiStat { } SSchApiStat; typedef struct SSchRuntimeStat { - #if defined(WINDOWS) || defined(_TD_DARWIN_64) size_t avoidCompilationErrors; #endif @@ -96,7 +94,6 @@ typedef struct SSchRuntimeStat { } SSchRuntimeStat; typedef struct SSchJobStat { - #if defined(WINDOWS) || defined(_TD_DARWIN_64) size_t avoidCompilationErrors; #endif @@ -104,17 +101,17 @@ typedef struct SSchJobStat { } SSchJobStat; typedef struct SSchStat { - SSchApiStat api; - SSchRuntimeStat runtime; - SSchJobStat job; + SSchApiStat api; + SSchRuntimeStat runtime; + SSchJobStat job; } SSchStat; typedef struct SSchResInfo { - SExecResult* execRes; - void** fetchRes; - schedulerExecFp execFp; - schedulerFetchFp fetchFp; - void* cbParam; + SExecResult *execRes; + void **fetchRes; + schedulerExecFp execFp; + schedulerFetchFp fetchFp; + void *cbParam; } SSchResInfo; typedef struct SSchOpEvent { @@ -123,9 +120,9 @@ typedef struct SSchOpEvent { SSchedulerReq *pReq; } SSchOpEvent; -typedef int32_t (*schStatusEnterFp)(void* pHandle, void* pParam); -typedef int32_t (*schStatusLeaveFp)(void* pHandle, void* pParam); -typedef int32_t (*schStatusEventFp)(void* pHandle, void* pParam, void* pEvent); +typedef int32_t (*schStatusEnterFp)(void *pHandle, void *pParam); +typedef int32_t (*schStatusLeaveFp)(void *pHandle, void *pParam); +typedef int32_t (*schStatusEventFp)(void *pHandle, void *pParam, void *pEvent); typedef struct SSchStatusFps { EJobTaskType status; @@ -142,16 +139,16 @@ typedef struct SSchedulerCfg { } SSchedulerCfg; typedef struct SSchedulerMgmt { - uint64_t taskId; // sequential taksId - uint64_t sId; // schedulerId - SSchedulerCfg cfg; - bool exit; - int32_t jobRef; - int32_t jobNum; - SSchStat stat; - SRWLatch hbLock; - SHashObj *hbConnections; - void *queryMgmt; + uint64_t taskId; // sequential taksId + uint64_t sId; // schedulerId + SSchedulerCfg cfg; + bool exit; + int32_t jobRef; + int32_t jobNum; + SSchStat stat; + SRWLatch hbLock; + SHashObj *hbConnections; + void *queryMgmt; } SSchedulerMgmt; typedef struct SSchCallbackParamHeader { @@ -174,11 +171,11 @@ typedef struct SSchHbCallbackParam { } SSchHbCallbackParam; typedef struct SSchFlowControl { - SRWLatch lock; - bool sorted; - int32_t tableNumSum; - uint32_t execTaskNum; - SArray *taskList; // Element is SSchTask* + SRWLatch lock; + bool sorted; + int32_t tableNumSum; + uint32_t execTaskNum; + SArray *taskList; // Element is SSchTask* } SSchFlowControl; typedef struct SSchNodeInfo { @@ -187,50 +184,50 @@ typedef struct SSchNodeInfo { } SSchNodeInfo; typedef struct SSchLevel { - int32_t level; - int8_t status; - SRWLatch lock; - int32_t taskFailed; - int32_t taskSucceed; - int32_t taskNum; - int32_t taskLaunchedNum; - int32_t taskDoneNum; - SArray *subTasks; // Element is SSchTask + int32_t level; + int8_t status; + SRWLatch lock; + int32_t taskFailed; + int32_t taskSucceed; + int32_t taskNum; + int32_t taskLaunchedNum; + int32_t taskDoneNum; + SArray *subTasks; // Element is SSchTask } SSchLevel; typedef struct SSchTaskProfile { - int64_t startTs; - SArray* execTime; - int64_t waitTime; - int64_t endTs; + int64_t startTs; + SArray *execTime; + int64_t waitTime; + int64_t endTs; } SSchTaskProfile; typedef struct SSchTask { - uint64_t taskId; // task id - SRWLatch lock; // task reentrant lock - int32_t maxExecTimes; // task max exec times - int32_t maxRetryTimes; // task max retry times - int32_t retryTimes; // task retry times - bool waitRetry; // wait for retry - int32_t execId; // task current execute index - SSchLevel *level; // level - SRWLatch planLock; // task update plan lock - SSubplan *plan; // subplan - char *msg; // operator tree - int32_t msgLen; // msg length - int8_t status; // task status - int32_t lastMsgType; // last sent msg type - int64_t timeoutUsec; // task timeout useconds before reschedule - SQueryNodeAddr succeedAddr; // task executed success node address - int8_t candidateIdx; // current try condidation index - SArray *candidateAddrs; // condidate node addresses, element is SQueryNodeAddr - SHashObj *execNodes; // all tried node for current task, element is SSchNodeInfo - SSchTaskProfile profile; // task execution profile - int32_t childReady; // child task ready number - SArray *children; // the datasource tasks,from which to fetch the result, element is SQueryTask* - SArray *parents; // the data destination tasks, get data from current task, element is SQueryTask* - void* handle; // task send handle - bool registerdHb; // registered in hb + uint64_t taskId; // task id + SRWLatch lock; // task reentrant lock + int32_t maxExecTimes; // task max exec times + int32_t maxRetryTimes; // task max retry times + int32_t retryTimes; // task retry times + bool waitRetry; // wait for retry + int32_t execId; // task current execute index + SSchLevel *level; // level + SRWLatch planLock; // task update plan lock + SSubplan *plan; // subplan + char *msg; // operator tree + int32_t msgLen; // msg length + int8_t status; // task status + int32_t lastMsgType; // last sent msg type + int64_t timeoutUsec; // task timeout useconds before reschedule + SQueryNodeAddr succeedAddr; // task executed success node address + int8_t candidateIdx; // current try condidation index + SArray *candidateAddrs; // condidate node addresses, element is SQueryNodeAddr + SHashObj *execNodes; // all tried node for current task, element is SSchNodeInfo + SSchTaskProfile profile; // task execution profile + int32_t childReady; // child task ready number + SArray *children; // the datasource tasks,from which to fetch the result, element is SQueryTask* + SArray *parents; // the data destination tasks, get data from current task, element is SQueryTask* + void *handle; // task send handle + bool registerdHb; // registered in hb } SSchTask; typedef struct SSchJobAttr { @@ -243,46 +240,46 @@ typedef struct SSchJobAttr { } SSchJobAttr; typedef struct { - int32_t op; - SRWLatch lock; - bool syncReq; + int32_t op; + SRWLatch lock; + bool syncReq; } SSchOpStatus; typedef struct SSchJob { - int64_t refId; - uint64_t queryId; - SSchJobAttr attr; - int32_t levelNum; - int32_t taskNum; - SRequestConnInfo conn; - SArray *nodeList; // qnode/vnode list, SArray - SArray *levels; // starting from 0. SArray - SQueryPlan *pDag; - int64_t allocatorRefId; + int64_t refId; + uint64_t queryId; + SSchJobAttr attr; + int32_t levelNum; + int32_t taskNum; + SRequestConnInfo conn; + SArray *nodeList; // qnode/vnode list, SArray + SArray *levels; // starting from 0. SArray + SQueryPlan *pDag; + int64_t allocatorRefId; - SArray *dataSrcTasks; // SArray - int32_t levelIdx; - SEpSet dataSrcEps; - SHashObj *taskList; - SHashObj *execTasks; // executing and executed tasks, key:taskid, value:SQueryTask* - SHashObj *flowCtrl; // key is ep, element is SSchFlowControl - - SExplainCtx *explainCtx; - int8_t status; - SQueryNodeAddr resNode; - tsem_t rspSem; - SSchOpStatus opStatus; - schedulerChkKillFp chkKillFp; - void* chkKillParam; - SSchTask *fetchTask; - int32_t errCode; - SRWLatch resLock; - SExecResult execRes; - void *fetchRes; //TODO free it or not - bool fetched; - int32_t resNumOfRows; - SSchResInfo userRes; - char *sql; + SArray *dataSrcTasks; // SArray + int32_t levelIdx; + SEpSet dataSrcEps; + SHashObj *taskList; + SHashObj *execTasks; // executing and executed tasks, key:taskid, value:SQueryTask* + SHashObj *flowCtrl; // key is ep, element is SSchFlowControl + + SExplainCtx *explainCtx; + int8_t status; + SQueryNodeAddr resNode; + tsem_t rspSem; + SSchOpStatus opStatus; + schedulerChkKillFp chkKillFp; + void *chkKillParam; + SSchTask *fetchTask; + int32_t errCode; + SRWLatch resLock; + SExecResult execRes; + void *fetchRes; // TODO free it or not + bool fetched; + int32_t resNumOfRows; + SSchResInfo userRes; + char *sql; SQueryProfileSummary summary; } SSchJob; @@ -294,226 +291,282 @@ typedef struct SSchTaskCtx { extern SSchedulerMgmt schMgmt; -#define SCH_TASK_TIMEOUT(_task) ((taosGetTimestampUs() - *(int64_t*)taosArrayGet((_task)->profile.execTime, (_task)->execId)) > (_task)->timeoutUsec) +#define SCH_TASK_TIMEOUT(_task) \ + ((taosGetTimestampUs() - *(int64_t *)taosArrayGet((_task)->profile.execTime, (_task)->execId)) > (_task)->timeoutUsec) #define SCH_TASK_READY_FOR_LAUNCH(readyNum, task) ((readyNum) >= taosArrayGetSize((task)->children)) -#define SCH_LOCK_TASK(_task) SCH_LOCK(SCH_WRITE, &(_task)->lock) +#define SCH_LOCK_TASK(_task) SCH_LOCK(SCH_WRITE, &(_task)->lock) #define SCH_UNLOCK_TASK(_task) SCH_UNLOCK(SCH_WRITE, &(_task)->lock) -#define SCH_TASK_ID(_task) ((_task) ? (_task)->taskId : -1) +#define SCH_TASK_ID(_task) ((_task) ? (_task)->taskId : -1) #define SCH_TASK_EID(_task) ((_task) ? (_task)->execId : -1) #define SCH_IS_DATA_BIND_QRY_TASK(task) ((task)->plan->subplanType == SUBPLAN_TYPE_SCAN) -#define SCH_IS_DATA_BIND_TASK(task) (((task)->plan->subplanType == SUBPLAN_TYPE_SCAN) || ((task)->plan->subplanType == SUBPLAN_TYPE_MODIFY)) +#define SCH_IS_DATA_BIND_TASK(task) \ + (((task)->plan->subplanType == SUBPLAN_TYPE_SCAN) || ((task)->plan->subplanType == SUBPLAN_TYPE_MODIFY)) #define SCH_IS_LEAF_TASK(_job, _task) (((_task)->level->level + 1) == (_job)->levelNum) -#define SCH_IS_DATA_MERGE_TASK(task) (!SCH_IS_DATA_BIND_TASK(task)) -#define SCH_IS_LOCAL_EXEC_TASK(_job, _task) ((_job)->attr.localExec && SCH_IS_QUERY_JOB(_job) && (!SCH_IS_INSERT_JOB(_job)) && (!SCH_IS_DATA_BIND_QRY_TASK(_task))) +#define SCH_IS_DATA_MERGE_TASK(task) (!SCH_IS_DATA_BIND_TASK(task)) +#define SCH_IS_LOCAL_EXEC_TASK(_job, _task) \ + ((_job)->attr.localExec && SCH_IS_QUERY_JOB(_job) && (!SCH_IS_INSERT_JOB(_job)) && \ + (!SCH_IS_DATA_BIND_QRY_TASK(_task))) #define SCH_SET_TASK_STATUS(task, st) atomic_store_8(&(task)->status, st) -#define SCH_GET_TASK_STATUS(task) atomic_load_8(&(task)->status) +#define SCH_GET_TASK_STATUS(task) atomic_load_8(&(task)->status) #define SCH_GET_TASK_STATUS_STR(task) jobTaskStatusStr(SCH_GET_TASK_STATUS(task)) -#define SCH_GET_TASK_HANDLE(_task) ((_task) ? (_task)->handle : NULL) +#define SCH_GET_TASK_HANDLE(_task) ((_task) ? (_task)->handle : NULL) #define SCH_SET_TASK_HANDLE(_task, _handle) ((_task)->handle = (_handle)) #define SCH_SET_JOB_STATUS(job, st) atomic_store_8(&(job)->status, st) -#define SCH_GET_JOB_STATUS(job) atomic_load_8(&(job)->status) +#define SCH_GET_JOB_STATUS(job) atomic_load_8(&(job)->status) #define SCH_GET_JOB_STATUS_STR(job) jobTaskStatusStr(SCH_GET_JOB_STATUS(job)) #define SCH_JOB_IN_SYNC_OP(job) ((job)->opStatus.op && (job)->opStatus.syncReq) -#define SCH_JOB_IN_ASYNC_EXEC_OP(job) ((SCH_OP_EXEC == atomic_val_compare_exchange_32(&(job)->opStatus.op, SCH_OP_EXEC, SCH_OP_NULL)) && (!(job)->opStatus.syncReq)) -#define SCH_JOB_IN_ASYNC_FETCH_OP(job) ((SCH_OP_FETCH == atomic_val_compare_exchange_32(&(job)->opStatus.op, SCH_OP_FETCH, SCH_OP_NULL)) && (!(job)->opStatus.syncReq)) +#define SCH_JOB_IN_ASYNC_EXEC_OP(job) \ + ((SCH_OP_EXEC == atomic_val_compare_exchange_32(&(job)->opStatus.op, SCH_OP_EXEC, SCH_OP_NULL)) && \ + (!(job)->opStatus.syncReq)) +#define SCH_JOB_IN_ASYNC_FETCH_OP(job) \ + ((SCH_OP_FETCH == atomic_val_compare_exchange_32(&(job)->opStatus.op, SCH_OP_FETCH, SCH_OP_NULL)) && \ + (!(job)->opStatus.syncReq)) #define SCH_SET_JOB_NEED_FLOW_CTRL(_job) (_job)->attr.needFlowCtrl = true -#define SCH_JOB_NEED_FLOW_CTRL(_job) ((_job)->attr.needFlowCtrl) -#define SCH_TASK_NEED_FLOW_CTRL(_job, _task) (SCH_IS_DATA_BIND_QRY_TASK(_task) && SCH_JOB_NEED_FLOW_CTRL(_job) && SCH_IS_LEVEL_UNFINISHED((_task)->level)) -#define SCH_FETCH_TYPE(_pSrcTask) (SCH_IS_DATA_BIND_QRY_TASK(_pSrcTask) ? TDMT_SCH_FETCH : TDMT_SCH_MERGE_FETCH) +#define SCH_JOB_NEED_FLOW_CTRL(_job) ((_job)->attr.needFlowCtrl) +#define SCH_TASK_NEED_FLOW_CTRL(_job, _task) \ + (SCH_IS_DATA_BIND_QRY_TASK(_task) && SCH_JOB_NEED_FLOW_CTRL(_job) && SCH_IS_LEVEL_UNFINISHED((_task)->level)) +#define SCH_FETCH_TYPE(_pSrcTask) (SCH_IS_DATA_BIND_QRY_TASK(_pSrcTask) ? TDMT_SCH_FETCH : TDMT_SCH_MERGE_FETCH) #define SCH_TASK_NEED_FETCH(_task) ((_task)->plan->subplanType != SUBPLAN_TYPE_MODIFY) -#define SCH_SET_JOB_TYPE(_job, type) do { if ((type) != SUBPLAN_TYPE_MODIFY) { (_job)->attr.queryJob = true; } else { (_job)->attr.insertJob = true; } } while (0) -#define SCH_IS_QUERY_JOB(_job) ((_job)->attr.queryJob) -#define SCH_IS_INSERT_JOB(_job) ((_job)->attr.insertJob) +#define SCH_SET_JOB_TYPE(_job, type) \ + do { \ + if ((type) != SUBPLAN_TYPE_MODIFY) { \ + (_job)->attr.queryJob = true; \ + } else { \ + (_job)->attr.insertJob = true; \ + } \ + } while (0) +#define SCH_IS_QUERY_JOB(_job) ((_job)->attr.queryJob) +#define SCH_IS_INSERT_JOB(_job) ((_job)->attr.insertJob) #define SCH_JOB_NEED_FETCH(_job) ((_job)->attr.needFetch) -#define SCH_JOB_NEED_WAIT(_job) (!SCH_IS_QUERY_JOB(_job)) -#define SCH_JOB_NEED_DROP(_job) (SCH_IS_QUERY_JOB(_job)) +#define SCH_JOB_NEED_WAIT(_job) (!SCH_IS_QUERY_JOB(_job)) +#define SCH_JOB_NEED_DROP(_job) (SCH_IS_QUERY_JOB(_job)) #define SCH_IS_EXPLAIN_JOB(_job) (EXPLAIN_MODE_ANALYZE == (_job)->attr.explainMode) -#define SCH_NETWORK_ERR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL) -#define SCH_MERGE_TASK_NETWORK_ERR(_task, _code, _len) (SCH_NETWORK_ERR(_code) && (((_len) > 0) || (!SCH_IS_DATA_BIND_TASK(_task)))) -#define SCH_REDIRECT_MSGTYPE(_msgType) ((_msgType) == TDMT_SCH_LINK_BROKEN || (_msgType) == TDMT_SCH_QUERY || (_msgType) == TDMT_SCH_MERGE_QUERY || (_msgType) == TDMT_SCH_FETCH || (_msgType) == TDMT_SCH_MERGE_FETCH) -#define SCH_TASK_NEED_REDIRECT(_task, _msgType, _code, _rspLen) (SCH_REDIRECT_MSGTYPE(_msgType) && (NEED_SCHEDULER_REDIRECT_ERROR(_code) || SCH_MERGE_TASK_NETWORK_ERR((_task), (_code), (_rspLen)))) -#define SCH_NEED_RETRY(_msgType, _code) ((SCH_NETWORK_ERR(_code) && SCH_REDIRECT_MSGTYPE(_msgType)) || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR) +#define SCH_NETWORK_ERR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL) +#define SCH_MERGE_TASK_NETWORK_ERR(_task, _code, _len) \ + (SCH_NETWORK_ERR(_code) && (((_len) > 0) || (!SCH_IS_DATA_BIND_TASK(_task)))) +#define SCH_REDIRECT_MSGTYPE(_msgType) \ + ((_msgType) == TDMT_SCH_LINK_BROKEN || (_msgType) == TDMT_SCH_QUERY || (_msgType) == TDMT_SCH_MERGE_QUERY || \ + (_msgType) == TDMT_SCH_FETCH || (_msgType) == TDMT_SCH_MERGE_FETCH) +#define SCH_TASK_NEED_REDIRECT(_task, _msgType, _code, _rspLen) \ + (SCH_REDIRECT_MSGTYPE(_msgType) && \ + (NEED_SCHEDULER_REDIRECT_ERROR(_code) || SCH_MERGE_TASK_NETWORK_ERR((_task), (_code), (_rspLen)))) +#define SCH_NEED_RETRY(_msgType, _code) \ + ((SCH_NETWORK_ERR(_code) && SCH_REDIRECT_MSGTYPE(_msgType)) || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR) #define SCH_IS_LEVEL_UNFINISHED(_level) ((_level)->taskLaunchedNum < (_level)->taskNum) -#define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse]) -#define SCH_SWITCH_EPSET(_addr) ((_addr)->epSet.inUse = ((_addr)->epSet.inUse + 1) % (_addr)->epSet.numOfEps) -#define SCH_TASK_NUM_OF_EPS(_addr) ((_addr)->epSet.numOfEps) +#define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse]) +#define SCH_SWITCH_EPSET(_addr) ((_addr)->epSet.inUse = ((_addr)->epSet.inUse + 1) % (_addr)->epSet.numOfEps) +#define SCH_TASK_NUM_OF_EPS(_addr) ((_addr)->epSet.numOfEps) -#define SCH_LOG_TASK_START_TS(_task) \ - do { \ - int64_t us = taosGetTimestampUs(); \ - taosArrayPush((_task)->profile.execTime, &us); \ - if (0 == (_task)->execId) { \ - (_task)->profile.startTs = us; \ - } \ - } while (0) +#define SCH_LOG_TASK_START_TS(_task) \ + do { \ + int64_t us = taosGetTimestampUs(); \ + taosArrayPush((_task)->profile.execTime, &us); \ + if (0 == (_task)->execId) { \ + (_task)->profile.startTs = us; \ + } \ + } while (0) -#define SCH_LOG_TASK_WAIT_TS(_task) \ - do { \ - int64_t us = taosGetTimestampUs(); \ - (_task)->profile.waitTime += us - *(int64_t*)taosArrayGet((_task)->profile.execTime, (_task)->execId); \ - } while (0) +#define SCH_LOG_TASK_WAIT_TS(_task) \ + do { \ + int64_t us = taosGetTimestampUs(); \ + (_task)->profile.waitTime += us - *(int64_t *)taosArrayGet((_task)->profile.execTime, (_task)->execId); \ + } while (0) - -#define SCH_LOG_TASK_END_TS(_task) \ - do { \ - int64_t us = taosGetTimestampUs(); \ - int32_t idx = (_task)->execId % (_task)->maxExecTimes; \ +#define SCH_LOG_TASK_END_TS(_task) \ + do { \ + int64_t us = taosGetTimestampUs(); \ + int32_t idx = (_task)->execId % (_task)->maxExecTimes; \ int64_t *startts = taosArrayGet((_task)->profile.execTime, (_task)->execId); \ - *startts = us - *startts; \ - (_task)->profile.endTs = us; \ - } while (0) - + *startts = us - *startts; \ + (_task)->profile.endTs = us; \ + } while (0) #define SCH_JOB_ELOG(param, ...) qError("QID:0x%" PRIx64 " " param, pJob->queryId, __VA_ARGS__) #define SCH_JOB_DLOG(param, ...) qDebug("QID:0x%" PRIx64 " " param, pJob->queryId, __VA_ARGS__) -#define SCH_TASK_ELOG(param, ...) \ - qError("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask),__VA_ARGS__) -#define SCH_TASK_DLOG(param, ...) \ - qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask),__VA_ARGS__) -#define SCH_TASK_TLOG(param, ...) \ - qTrace("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask),__VA_ARGS__) -#define SCH_TASK_DLOGL(param, ...) \ - qDebugL("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask),__VA_ARGS__) -#define SCH_TASK_WLOG(param, ...) \ - qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask),__VA_ARGS__) +#define SCH_TASK_ELOG(param, ...) \ + qError("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask), \ + __VA_ARGS__) +#define SCH_TASK_DLOG(param, ...) \ + qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask), \ + __VA_ARGS__) +#define SCH_TASK_TLOG(param, ...) \ + qTrace("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask), \ + __VA_ARGS__) +#define SCH_TASK_DLOGL(param, ...) \ + qDebugL("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask), \ + __VA_ARGS__) +#define SCH_TASK_WLOG(param, ...) \ + qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d " param, pJob->queryId, SCH_TASK_ID(pTask), SCH_TASK_EID(pTask), \ + __VA_ARGS__) -#define SCH_SET_ERRNO(_err) do { if (TSDB_CODE_SCH_IGNORE_ERROR != (_err)) { terrno = (_err); } } while (0) -#define SCH_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { SCH_SET_ERRNO(_code); return _code; } } while (0) -#define SCH_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { SCH_SET_ERRNO(_code); } return _code; } while (0) -#define SCH_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { SCH_SET_ERRNO(code); goto _return; } } while (0) +#define SCH_SET_ERRNO(_err) \ + do { \ + if (TSDB_CODE_SCH_IGNORE_ERROR != (_err)) { \ + terrno = (_err); \ + } \ + } while (0) +#define SCH_ERR_RET(c) \ + do { \ + int32_t _code = c; \ + if (_code != TSDB_CODE_SUCCESS) { \ + SCH_SET_ERRNO(_code); \ + return _code; \ + } \ + } while (0) +#define SCH_RET(c) \ + do { \ + int32_t _code = c; \ + if (_code != TSDB_CODE_SUCCESS) { \ + SCH_SET_ERRNO(_code); \ + } \ + return _code; \ + } while (0) +#define SCH_ERR_JRET(c) \ + do { \ + code = c; \ + if (code != TSDB_CODE_SUCCESS) { \ + SCH_SET_ERRNO(code); \ + goto _return; \ + } \ + } while (0) -#define SCH_LOCK_DEBUG(...) do { if (gSCHDebug.lockEnable) { qDebug(__VA_ARGS__); } } while (0) +#define SCH_LOCK_DEBUG(...) \ + do { \ + if (gSCHDebug.lockEnable) { \ + qDebug(__VA_ARGS__); \ + } \ + } while (0) #define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000 -#define SCH_LOCK(type, _lock) do { \ - if (SCH_READ == (type)) { \ - assert(atomic_load_32(_lock) >= 0); \ - SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosRLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - assert(atomic_load_32(_lock) > 0); \ - } else { \ - assert(atomic_load_32(_lock) >= 0); \ - SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosWLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - assert(atomic_load_32(_lock) == TD_RWLATCH_WRITE_FLAG_COPY); \ - } \ -} while (0) +#define SCH_LOCK(type, _lock) \ + do { \ + if (SCH_READ == (type)) { \ + assert(atomic_load_32(_lock) >= 0); \ + SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosRLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + assert(atomic_load_32(_lock) > 0); \ + } else { \ + assert(atomic_load_32(_lock) >= 0); \ + SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosWLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + assert(atomic_load_32(_lock) == TD_RWLATCH_WRITE_FLAG_COPY); \ + } \ + } while (0) -#define SCH_UNLOCK(type, _lock) do { \ - if (SCH_READ == (type)) { \ - assert(atomic_load_32((_lock)) > 0); \ - SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosRUnLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - assert(atomic_load_32((_lock)) >= 0); \ - } else { \ - assert(atomic_load_32((_lock)) & TD_RWLATCH_WRITE_FLAG_COPY); \ - SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosWUnLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - assert(atomic_load_32((_lock)) >= 0); \ - } \ -} while (0) +#define SCH_UNLOCK(type, _lock) \ + do { \ + if (SCH_READ == (type)) { \ + assert(atomic_load_32((_lock)) > 0); \ + SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosRUnLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + assert(atomic_load_32((_lock)) >= 0); \ + } else { \ + assert(atomic_load_32((_lock)) & TD_RWLATCH_WRITE_FLAG_COPY); \ + SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosWUnLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + assert(atomic_load_32((_lock)) >= 0); \ + } \ + } while (0) - -void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask); -void schCleanClusterHb(void* pTrans); -int32_t schLaunchTask(SSchJob *job, SSchTask *task); -int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType); +void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask); +void schCleanClusterHb(void *pTrans); +int32_t schLaunchTask(SSchJob *job, SSchTask *task); +int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType); SSchJob *schAcquireJob(int64_t refId); -int32_t schReleaseJob(int64_t refId); -void schFreeFlowCtrl(SSchJob *pJob); -int32_t schChkJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel); -int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask); -int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough); -int32_t schLaunchTasksInFlowCtrlList(SSchJob *pJob, SSchTask *pTask); -int32_t schAsyncLaunchTaskImpl(SSchJob *pJob, SSchTask *pTask); -int32_t schLaunchFetchTask(SSchJob *pJob); -int32_t schProcessOnTaskFailure(SSchJob *pJob, SSchTask *pTask, int32_t errCode); -int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray* taskAction); -int32_t schCloneSMsgSendInfo(void *src, void **dst); -int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob); -void schFreeJobImpl(void *job); -int32_t schMakeHbRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx); -int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask); -int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans); -int32_t schHandleHbCallback(void *param, SDataBuf *pMsg, int32_t code); -void schFreeRpcCtx(SRpcCtx *pCtx); -int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp); -bool schJobNeedToStop(SSchJob *pJob, int8_t *pStatus); -int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask); -int32_t schSaveJobExecRes(SSchJob *pJob, SQueryTableRsp *rsp); -int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRsp *pRsp); -void schProcessOnDataFetched(SSchJob *job); -int32_t schGetTaskInJob(SSchJob *pJob, uint64_t taskId, SSchTask **pTask); -void schFreeRpcCtxVal(const void *arg); -int32_t schMakeBrokenLinkVal(SSchJob *pJob, SSchTask *pTask, SRpcBrokenlinkVal *brokenVal, bool isHb); -int32_t schAppendTaskExecNode(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, int32_t execId); -int32_t schExecStaticExplainJob(SSchedulerReq *pReq, int64_t *job, bool sync); -int32_t schUpdateJobStatus(SSchJob *pJob, int8_t newStatus); -int32_t schCancelJob(SSchJob *pJob); -int32_t schProcessOnJobDropped(SSchJob *pJob, int32_t errCode); +int32_t schReleaseJob(int64_t refId); +void schFreeFlowCtrl(SSchJob *pJob); +int32_t schChkJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel); +int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask); +int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough); +int32_t schLaunchTasksInFlowCtrlList(SSchJob *pJob, SSchTask *pTask); +int32_t schAsyncLaunchTaskImpl(SSchJob *pJob, SSchTask *pTask); +int32_t schLaunchFetchTask(SSchJob *pJob); +int32_t schProcessOnTaskFailure(SSchJob *pJob, SSchTask *pTask, int32_t errCode); +int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray *taskAction); +int32_t schCloneSMsgSendInfo(void *src, void **dst); +int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob); +void schFreeJobImpl(void *job); +int32_t schMakeHbRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx); +int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask); +int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans); +int32_t schHandleHbCallback(void *param, SDataBuf *pMsg, int32_t code); +void schFreeRpcCtx(SRpcCtx *pCtx); +int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp); +bool schJobNeedToStop(SSchJob *pJob, int8_t *pStatus); +int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask); +int32_t schSaveJobExecRes(SSchJob *pJob, SQueryTableRsp *rsp); +int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRsp *pRsp); +void schProcessOnDataFetched(SSchJob *job); +int32_t schGetTaskInJob(SSchJob *pJob, uint64_t taskId, SSchTask **pTask); +void schFreeRpcCtxVal(const void *arg); +int32_t schMakeBrokenLinkVal(SSchJob *pJob, SSchTask *pTask, SRpcBrokenlinkVal *brokenVal, bool isHb); +int32_t schAppendTaskExecNode(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, int32_t execId); +int32_t schExecStaticExplainJob(SSchedulerReq *pReq, int64_t *job, bool sync); +int32_t schUpdateJobStatus(SSchJob *pJob, int8_t newStatus); +int32_t schCancelJob(SSchJob *pJob); +int32_t schProcessOnJobDropped(SSchJob *pJob, int32_t errCode); uint64_t schGenTaskId(void); -void schCloseJobRef(void); -int32_t schAsyncExecJob(SSchedulerReq *pReq, int64_t *pJob); -int32_t schJobFetchRows(SSchJob *pJob); -int32_t schJobFetchRowsA(SSchJob *pJob); -int32_t schUpdateTaskHandle(SSchJob *pJob, SSchTask *pTask, bool dropExecNode, void *handle, int32_t execId); -int32_t schProcessOnTaskStatusRsp(SQueryNodeEpId* pEpId, SArray* pStatusList); -char* schGetOpStr(SCH_OP_TYPE type); -int32_t schBeginOperation(SSchJob *pJob, SCH_OP_TYPE type, bool sync); -int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq); -int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq); -int32_t schDumpJobExecRes(SSchJob* pJob, SExecResult* pRes); -int32_t schUpdateTaskCandidateAddr(SSchJob *pJob, SSchTask *pTask, SEpSet* pEpSet); -int32_t schHandleRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf* pData, int32_t rspCode); -void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int32_t errCode); -int32_t schProcessOnOpBegin(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq); -void schProcessOnCbEnd(SSchJob *pJob, SSchTask *pTask, int32_t errCode); -int32_t schProcessOnCbBegin(SSchJob** job, SSchTask** task, uint64_t qId, int64_t rId, uint64_t tId); -void schDropTaskOnExecNode(SSchJob *pJob, SSchTask *pTask); -bool schJobDone(SSchJob *pJob); -int32_t schRemoveTaskFromExecList(SSchJob *pJob, SSchTask *pTask); -int32_t schLaunchJobLowerLevel(SSchJob *pJob, SSchTask *pTask); -int32_t schSwitchJobStatus(SSchJob* pJob, int32_t status, void* param); -int32_t schHandleOpBeginEvent(int64_t jobId, SSchJob** job, SCH_OP_TYPE type, SSchedulerReq* pReq); -int32_t schHandleOpEndEvent(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int32_t errCode); -int32_t schHandleTaskRetry(SSchJob *pJob, SSchTask *pTask); -void schUpdateJobErrCode(SSchJob *pJob, int32_t errCode); -int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bool *needRetry); -int32_t schProcessOnJobFailure(SSchJob *pJob, int32_t errCode); -int32_t schProcessOnJobPartialSuccess(SSchJob *pJob); -void schFreeTask(SSchJob *pJob, SSchTask *pTask); -void schDropTaskInHashList(SSchJob *pJob, SHashObj *list); -int32_t schLaunchLevelTasks(SSchJob *pJob, SSchLevel *level); -int32_t schGetTaskFromList(SHashObj *pTaskList, uint64_t taskId, SSchTask **pTask); -int32_t schInitTask(SSchJob *pJob, SSchTask *pTask, SSubplan *pPlan, SSchLevel *pLevel); -int32_t schSwitchTaskCandidateAddr(SSchJob *pJob, SSchTask *pTask); -void schDirectPostJobRes(SSchedulerReq* pReq, int32_t errCode); -int32_t schHandleJobFailure(SSchJob *pJob, int32_t errCode); -int32_t schHandleJobDrop(SSchJob *pJob, int32_t errCode); -bool schChkCurrentOp(SSchJob *pJob, int32_t op, int8_t sync); -int32_t schProcessFetchRsp(SSchJob *pJob, SSchTask *pTask, char *msg, int32_t rspCode); -int32_t schProcessExplainRsp(SSchJob *pJob, SSchTask *pTask, SExplainRsp *rsp); +void schCloseJobRef(void); +int32_t schAsyncExecJob(SSchedulerReq *pReq, int64_t *pJob); +int32_t schJobFetchRows(SSchJob *pJob); +int32_t schJobFetchRowsA(SSchJob *pJob); +int32_t schUpdateTaskHandle(SSchJob *pJob, SSchTask *pTask, bool dropExecNode, void *handle, int32_t execId); +int32_t schProcessOnTaskStatusRsp(SQueryNodeEpId *pEpId, SArray *pStatusList); +char *schGetOpStr(SCH_OP_TYPE type); +int32_t schBeginOperation(SSchJob *pJob, SCH_OP_TYPE type, bool sync); +int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq); +int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq); +int32_t schDumpJobExecRes(SSchJob *pJob, SExecResult *pRes); +int32_t schUpdateTaskCandidateAddr(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet); +int32_t schHandleRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32_t rspCode); +void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq *pReq, int32_t errCode); +int32_t schProcessOnOpBegin(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq *pReq); +void schProcessOnCbEnd(SSchJob *pJob, SSchTask *pTask, int32_t errCode); +int32_t schProcessOnCbBegin(SSchJob **job, SSchTask **task, uint64_t qId, int64_t rId, uint64_t tId); +void schDropTaskOnExecNode(SSchJob *pJob, SSchTask *pTask); +bool schJobDone(SSchJob *pJob); +int32_t schRemoveTaskFromExecList(SSchJob *pJob, SSchTask *pTask); +int32_t schLaunchJobLowerLevel(SSchJob *pJob, SSchTask *pTask); +int32_t schSwitchJobStatus(SSchJob *pJob, int32_t status, void *param); +int32_t schHandleOpBeginEvent(int64_t jobId, SSchJob **job, SCH_OP_TYPE type, SSchedulerReq *pReq); +int32_t schHandleOpEndEvent(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq *pReq, int32_t errCode); +int32_t schHandleTaskRetry(SSchJob *pJob, SSchTask *pTask); +void schUpdateJobErrCode(SSchJob *pJob, int32_t errCode); +int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bool *needRetry); +int32_t schProcessOnJobFailure(SSchJob *pJob, int32_t errCode); +int32_t schProcessOnJobPartialSuccess(SSchJob *pJob); +void schFreeTask(SSchJob *pJob, SSchTask *pTask); +void schDropTaskInHashList(SSchJob *pJob, SHashObj *list); +int32_t schLaunchLevelTasks(SSchJob *pJob, SSchLevel *level); +int32_t schGetTaskFromList(SHashObj *pTaskList, uint64_t taskId, SSchTask **pTask); +int32_t schInitTask(SSchJob *pJob, SSchTask *pTask, SSubplan *pPlan, SSchLevel *pLevel); +int32_t schSwitchTaskCandidateAddr(SSchJob *pJob, SSchTask *pTask); +void schDirectPostJobRes(SSchedulerReq *pReq, int32_t errCode); +int32_t schHandleJobFailure(SSchJob *pJob, int32_t errCode); +int32_t schHandleJobDrop(SSchJob *pJob, int32_t errCode); +bool schChkCurrentOp(SSchJob *pJob, int32_t op, int8_t sync); +int32_t schProcessFetchRsp(SSchJob *pJob, SSchTask *pTask, char *msg, int32_t rspCode); +int32_t schProcessExplainRsp(SSchJob *pJob, SSchTask *pTask, SExplainRsp *rsp); extern SSchDebug gSCHDebug; - #ifdef __cplusplus } #endif diff --git a/source/libs/scheduler/src/schDbg.c b/source/libs/scheduler/src/schDbg.c index a6398522d3..d6c2b638b8 100644 --- a/source/libs/scheduler/src/schDbg.c +++ b/source/libs/scheduler/src/schDbg.c @@ -16,19 +16,17 @@ #include "query.h" #include "schInt.h" -tsem_t schdRspSem; +tsem_t schdRspSem; SSchDebug gSCHDebug = {0}; void schdExecCallback(SExecResult* pResult, void* param, int32_t code) { if (code) { pResult->code = code; } - + *(SExecResult*)param = *pResult; taosMemoryFree(pResult); tsem_post(&schdRspSem); } - - diff --git a/source/libs/scheduler/src/schFlowCtrl.c b/source/libs/scheduler/src/schFlowCtrl.c index c5c2bfb2bb..8980d08e89 100644 --- a/source/libs/scheduler/src/schFlowCtrl.c +++ b/source/libs/scheduler/src/schFlowCtrl.c @@ -13,10 +13,10 @@ * along with this program. If not, see . */ +#include "catalog.h" +#include "query.h" #include "schInt.h" #include "tmsg.h" -#include "query.h" -#include "catalog.h" #include "tref.h" void schFreeFlowCtrl(SSchJob *pJob) { @@ -25,14 +25,14 @@ void schFreeFlowCtrl(SSchJob *pJob) { } SSchFlowControl *ctrl = NULL; - void *pIter = taosHashIterate(pJob->flowCtrl, NULL); + void *pIter = taosHashIterate(pJob->flowCtrl, NULL); while (pIter) { ctrl = (SSchFlowControl *)pIter; if (ctrl->taskList) { taosArrayDestroy(ctrl->taskList); } - + pIter = taosHashIterate(pJob->flowCtrl, pIter); } @@ -59,7 +59,8 @@ int32_t schChkJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) { return TSDB_CODE_SUCCESS; } - pJob->flowCtrl = taosHashInit(pJob->taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK); + pJob->flowCtrl = + taosHashInit(pJob->taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK); if (NULL == pJob->flowCtrl) { SCH_JOB_ELOG("taosHashInit %d flowCtrl failed", pJob->taskNum); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); @@ -73,17 +74,17 @@ int32_t schChkJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) { } int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask) { - SSchLevel *pLevel = pTask->level; + SSchLevel *pLevel = pTask->level; SSchFlowControl *ctrl = NULL; - int32_t code = 0; - SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); - + int32_t code = 0; + SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); + ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp)); if (NULL == ctrl) { SCH_TASK_ELOG("taosHashGet node from flowCtrl failed, fqdn:%s, port:%d", ep->fqdn, ep->port); SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR); } - + SCH_LOCK(SCH_WRITE, &ctrl->lock); if (ctrl->execTaskNum <= 0) { SCH_TASK_ELOG("taosHashGet node from flowCtrl failed, fqdn:%s, port:%d", ep->fqdn, ep->port); @@ -93,8 +94,8 @@ int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask) { --ctrl->execTaskNum; ctrl->tableNumSum -= pTask->plan->execNodeStat.tableNum; - SCH_TASK_DLOG("task quota removed, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", - ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); + SCH_TASK_DLOG("task quota removed, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", ep->fqdn, + ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); _return: @@ -104,11 +105,11 @@ _return: } int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough) { - SSchLevel *pLevel = pTask->level; - int32_t code = 0; + SSchLevel *pLevel = pTask->level; + int32_t code = 0; SSchFlowControl *ctrl = NULL; - SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); - + SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); + do { ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp)); if (NULL == ctrl) { @@ -119,34 +120,34 @@ int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough) { if (HASH_NODE_EXIST(code)) { continue; } - + SCH_TASK_ELOG("taosHashPut flowCtrl failed, size:%d", (int32_t)sizeof(nctrl)); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); } - SCH_TASK_DLOG("task quota added, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", - ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, nctrl.tableNumSum, nctrl.execTaskNum); + SCH_TASK_DLOG("task quota added, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", ep->fqdn, + ep->port, pTask->plan->execNodeStat.tableNum, nctrl.tableNumSum, nctrl.execTaskNum); *enough = true; return TSDB_CODE_SUCCESS; } SCH_LOCK(SCH_WRITE, &ctrl->lock); - + if (0 == ctrl->execTaskNum) { ctrl->tableNumSum = pTask->plan->execNodeStat.tableNum; ++ctrl->execTaskNum; - + *enough = true; break; } - + int32_t sum = pTask->plan->execNodeStat.tableNum + ctrl->tableNumSum; - + if (sum <= schMgmt.cfg.maxNodeTableNum) { ctrl->tableNumSum = sum; ++ctrl->execTaskNum; - + *enough = true; break; } @@ -166,24 +167,25 @@ int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough) { *enough = false; ctrl->sorted = false; - + break; } while (true); _return: - SCH_TASK_DLOG("task quota %s added, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", - ((*enough)?"":"NOT"), ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); + SCH_TASK_DLOG("task quota %s added, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", + ((*enough) ? "" : "NOT"), ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, + ctrl->execTaskNum); SCH_UNLOCK(SCH_WRITE, &ctrl->lock); - + SCH_RET(code); } -int32_t schTaskTableNumCompare(const void* key1, const void* key2) { +int32_t schTaskTableNumCompare(const void *key1, const void *key2) { SSchTask *pTask1 = *(SSchTask **)key1; SSchTask *pTask2 = *(SSchTask **)key2; - + if (pTask1->plan->execNodeStat.tableNum < pTask2->plan->execNodeStat.tableNum) { return 1; } else if (pTask1->plan->execNodeStat.tableNum > pTask2->plan->execNodeStat.tableNum) { @@ -193,22 +195,21 @@ int32_t schTaskTableNumCompare(const void* key1, const void* key2) { } } - int32_t schLaunchTasksInFlowCtrlListImpl(SSchJob *pJob, SSchFlowControl *ctrl) { SCH_LOCK(SCH_WRITE, &ctrl->lock); - + if (NULL == ctrl->taskList || taosArrayGetSize(ctrl->taskList) <= 0) { SCH_UNLOCK(SCH_WRITE, &ctrl->lock); return TSDB_CODE_SUCCESS; } - int32_t remainNum = schMgmt.cfg.maxNodeTableNum - ctrl->tableNumSum; - int32_t taskNum = taosArrayGetSize(ctrl->taskList); - int32_t code = 0; + int32_t remainNum = schMgmt.cfg.maxNodeTableNum - ctrl->tableNumSum; + int32_t taskNum = taosArrayGetSize(ctrl->taskList); + int32_t code = 0; SSchTask *pTask = NULL; - + if (taskNum > 1 && !ctrl->sorted) { - taosArraySort(ctrl->taskList, schTaskTableNumCompare); // desc order + taosArraySort(ctrl->taskList, schTaskTableNumCompare); // desc order } for (int32_t i = 0; i < taskNum; ++i) { @@ -216,36 +217,36 @@ int32_t schLaunchTasksInFlowCtrlListImpl(SSchJob *pJob, SSchFlowControl *ctrl) { SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); if (pTask->plan->execNodeStat.tableNum > remainNum && ctrl->execTaskNum > 0) { - SCH_TASK_DLOG("task NOT to launch, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", - ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); + SCH_TASK_DLOG("task NOT to launch, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", ep->fqdn, + ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); continue; } - + ctrl->tableNumSum += pTask->plan->execNodeStat.tableNum; ++ctrl->execTaskNum; taosArrayRemove(ctrl->taskList, i); - - SCH_TASK_DLOG("task to launch, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", - ep->fqdn, ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); - + + SCH_TASK_DLOG("task to launch, fqdn:%s, port:%d, tableNum:%d, remainNum:%d, remainExecTaskNum:%d", ep->fqdn, + ep->port, pTask->plan->execNodeStat.tableNum, ctrl->tableNumSum, ctrl->execTaskNum); + SCH_ERR_JRET(schAsyncLaunchTaskImpl(pJob, pTask)); - + remainNum -= pTask->plan->execNodeStat.tableNum; if (remainNum <= 0) { - SCH_TASK_DLOG("no more task to launch, fqdn:%s, port:%d, remainNum:%d, remainExecTaskNum:%d", - ep->fqdn, ep->port, ctrl->tableNumSum, ctrl->execTaskNum); - + SCH_TASK_DLOG("no more task to launch, fqdn:%s, port:%d, remainNum:%d, remainExecTaskNum:%d", ep->fqdn, ep->port, + ctrl->tableNumSum, ctrl->execTaskNum); + break; } if (i < (taskNum - 1)) { SSchTask *pLastTask = *(SSchTask **)taosArrayGetLast(ctrl->taskList); if (remainNum < pLastTask->plan->execNodeStat.tableNum) { - SCH_TASK_DLOG("no more task to launch, fqdn:%s, port:%d, remainNum:%d, remainExecTaskNum:%d, smallestInList:%d", - ep->fqdn, ep->port, ctrl->tableNumSum, ctrl->execTaskNum, pLastTask->plan->execNodeStat.tableNum); - + SCH_TASK_DLOG("no more task to launch, fqdn:%s, port:%d, remainNum:%d, remainExecTaskNum:%d, smallestInList:%d", + ep->fqdn, ep->port, ctrl->tableNumSum, ctrl->execTaskNum, pLastTask->plan->execNodeStat.tableNum); + break; } } @@ -253,7 +254,7 @@ int32_t schLaunchTasksInFlowCtrlListImpl(SSchJob *pJob, SSchFlowControl *ctrl) { --i; --taskNum; } - + _return: SCH_UNLOCK(SCH_WRITE, &ctrl->lock); @@ -261,11 +262,10 @@ _return: if (code) { code = schProcessOnTaskFailure(pJob, pTask, code); } - + SCH_RET(code); } - int32_t schLaunchTasksInFlowCtrlList(SSchJob *pJob, SSchTask *pTask) { if (!SCH_TASK_NEED_FLOW_CTRL(pJob, pTask)) { return TSDB_CODE_SUCCESS; @@ -274,17 +274,16 @@ int32_t schLaunchTasksInFlowCtrlList(SSchJob *pJob, SSchTask *pTask) { SCH_ERR_RET(schDecTaskFlowQuota(pJob, pTask)); SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode); - + SSchFlowControl *ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp)); if (NULL == ctrl) { SCH_TASK_ELOG("taosHashGet node from flowCtrl failed, fqdn:%s, port:%d", ep->fqdn, ep->port); SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR); } - - int32_t code = schLaunchTasksInFlowCtrlListImpl(pJob, ctrl);; + + int32_t code = schLaunchTasksInFlowCtrlListImpl(pJob, ctrl); + ; SCH_ERR_RET(code); - - return code; // to avoid compiler error + + return code; // to avoid compiler error } - - diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c index 69495c8b7a..bb8451ba37 100644 --- a/source/libs/scheduler/src/schJob.c +++ b/source/libs/scheduler/src/schJob.c @@ -52,9 +52,8 @@ _return: bool schJobDone(SSchJob *pJob) { int8_t status = SCH_GET_JOB_STATUS(pJob); - - return (status == JOB_TASK_STATUS_FAIL || status == JOB_TASK_STATUS_DROP || - status == JOB_TASK_STATUS_SUCC); + + return (status == JOB_TASK_STATUS_FAIL || status == JOB_TASK_STATUS_DROP || status == JOB_TASK_STATUS_SUCC); } FORCE_INLINE bool schJobNeedToStop(SSchJob *pJob, int8_t *pStatus) { @@ -221,7 +220,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) { SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); } - SCH_TASK_DLOG("parents info, the %d parent TID 0x%" PRIx64, n, (*parentTask)->taskId); + SCH_TASK_DLOG("parents info, the %d parent TID 0x%" PRIx64, n, (*parentTask)->taskId); } SCH_TASK_DLOG("level:%d, parentNum:%d, childNum:%d", i, parentNum, childNum); @@ -235,7 +234,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) { SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR); } - SSchTask* pTask = taosArrayGet(pLevel->subTasks, 0); + SSchTask *pTask = taosArrayGet(pLevel->subTasks, 0); if (SUBPLAN_TYPE_MODIFY != pTask->plan->subplanType) { pJob->attr.needFetch = true; } @@ -244,7 +243,6 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) { return TSDB_CODE_SUCCESS; } - int32_t schAppendJobDataSrc(SSchJob *pJob, SSchTask *pTask) { if (!SCH_IS_DATA_BIND_QRY_TASK(pTask)) { return TSDB_CODE_SUCCESS; @@ -255,7 +253,6 @@ int32_t schAppendJobDataSrc(SSchJob *pJob, SSchTask *pTask) { return TSDB_CODE_SUCCESS; } - int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) { int32_t code = 0; pJob->queryId = pDag->queryId; @@ -365,7 +362,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) { SCH_ERR_JRET(schBuildTaskRalation(pJob, planToTask)); _return: - + if (planToTask) { taosHashCleanup(planToTask); } @@ -373,8 +370,7 @@ _return: SCH_RET(code); } - -int32_t schDumpJobExecRes(SSchJob* pJob, SExecResult* pRes) { +int32_t schDumpJobExecRes(SSchJob *pJob, SExecResult *pRes) { pRes->code = atomic_load_32(&pJob->errCode); pRes->numOfRows = pJob->resNumOfRows; pRes->res = pJob->execRes.res; @@ -387,13 +383,13 @@ int32_t schDumpJobExecRes(SSchJob* pJob, SExecResult* pRes) { return TSDB_CODE_SUCCESS; } -int32_t schDumpJobFetchRes(SSchJob* pJob, void** pData) { +int32_t schDumpJobFetchRes(SSchJob *pJob, void **pData) { int32_t code = 0; - + SCH_LOCK(SCH_WRITE, &pJob->resLock); pJob->fetched = true; - + if (pJob->fetchRes && ((SRetrieveTableRsp *)pJob->fetchRes)->completed) { SCH_ERR_JRET(schSwitchJobStatus(pJob, JOB_TASK_STATUS_SUCC, NULL)); } @@ -422,12 +418,12 @@ int32_t schDumpJobFetchRes(SSchJob* pJob, void** pData) { _return: SCH_UNLOCK(SCH_WRITE, &pJob->resLock); - + return code; } -int32_t schNotifyUserExecRes(SSchJob* pJob) { - SExecResult* pRes = taosMemoryCalloc(1, sizeof(SExecResult)); +int32_t schNotifyUserExecRes(SSchJob *pJob) { + SExecResult *pRes = taosMemoryCalloc(1, sizeof(SExecResult)); if (pRes) { schDumpJobExecRes(pJob, pRes); } @@ -439,9 +435,9 @@ int32_t schNotifyUserExecRes(SSchJob* pJob) { return TSDB_CODE_SUCCESS; } -int32_t schNotifyUserFetchRes(SSchJob* pJob) { - void* pRes = NULL; - +int32_t schNotifyUserFetchRes(SSchJob *pJob) { + void *pRes = NULL; + schDumpJobFetchRes(pJob, &pRes); SCH_JOB_DLOG("sch start to invoke fetch cb, code: %s", tstrerror(pJob->errCode)); @@ -453,17 +449,17 @@ int32_t schNotifyUserFetchRes(SSchJob* pJob) { void schPostJobRes(SSchJob *pJob, SCH_OP_TYPE op) { SCH_LOCK(SCH_WRITE, &pJob->opStatus.lock); - + if (SCH_OP_NULL == pJob->opStatus.op) { SCH_JOB_DLOG("job not in any operation, no need to post job res, status:%s", jobTaskStatusStr(pJob->status)); goto _return; } - + if (op && pJob->opStatus.op != op) { SCH_JOB_ELOG("job in operation %s mis-match with expected %s", schGetOpStr(pJob->opStatus.op), schGetOpStr(op)); goto _return; } - + if (SCH_JOB_IN_SYNC_OP(pJob)) { SCH_UNLOCK(SCH_WRITE, &pJob->opStatus.lock); tsem_post(&pJob->rspSem); @@ -487,7 +483,7 @@ _return: int32_t schProcessOnJobFailure(SSchJob *pJob, int32_t errCode) { schUpdateJobErrCode(pJob, errCode); - + int32_t code = atomic_load_32(&pJob->errCode); if (code) { SCH_JOB_DLOG("job failed with error %s", tstrerror(code)); @@ -507,9 +503,7 @@ int32_t schHandleJobFailure(SSchJob *pJob, int32_t errCode) { return TSDB_CODE_SCH_IGNORE_ERROR; } -int32_t schProcessOnJobDropped(SSchJob *pJob, int32_t errCode) { - SCH_RET(schProcessOnJobFailure(pJob, errCode)); -} +int32_t schProcessOnJobDropped(SSchJob *pJob, int32_t errCode) { SCH_RET(schProcessOnJobFailure(pJob, errCode)); } int32_t schHandleJobDrop(SSchJob *pJob, int32_t errCode) { if (TSDB_CODE_SCH_IGNORE_ERROR == errCode) { @@ -520,8 +514,7 @@ int32_t schHandleJobDrop(SSchJob *pJob, int32_t errCode) { return TSDB_CODE_SCH_IGNORE_ERROR; } - -int32_t schProcessOnJobPartialSuccess(SSchJob *pJob) { +int32_t schProcessOnJobPartialSuccess(SSchJob *pJob) { if (schChkCurrentOp(pJob, SCH_OP_FETCH, -1)) { SCH_ERR_RET(schLaunchFetchTask(pJob)); } else { @@ -531,9 +524,7 @@ int32_t schProcessOnJobPartialSuccess(SSchJob *pJob) { return TSDB_CODE_SUCCESS; } -void schProcessOnDataFetched(SSchJob *pJob) { - schPostJobRes(pJob, SCH_OP_FETCH); -} +void schProcessOnDataFetched(SSchJob *pJob) { schPostJobRes(pJob, SCH_OP_FETCH); } int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRsp *pRsp) { SCH_TASK_DLOG("got explain rsp, rows:%d, complete:%d", htonl(pRsp->numOfRows), pRsp->completed); @@ -548,14 +539,13 @@ int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRs return TSDB_CODE_SUCCESS; } - int32_t schLaunchJobLowerLevel(SSchJob *pJob, SSchTask *pTask) { if (!SCH_IS_QUERY_JOB(pJob)) { return TSDB_CODE_SUCCESS; } SSchLevel *pLevel = pTask->level; - int32_t doneNum = atomic_add_fetch_32(&pLevel->taskDoneNum, 1); + int32_t doneNum = atomic_add_fetch_32(&pLevel->taskDoneNum, 1); if (doneNum == pLevel->taskNum) { pJob->levelIdx--; @@ -566,7 +556,7 @@ int32_t schLaunchJobLowerLevel(SSchJob *pJob, SSchTask *pTask) { if (pTask->children && taosArrayGetSize(pTask->children) > 0) { continue; } - + SCH_ERR_RET(schLaunchTask(pJob, pTask)); } } @@ -577,11 +567,11 @@ int32_t schLaunchJobLowerLevel(SSchJob *pJob, SSchTask *pTask) { int32_t schSaveJobExecRes(SSchJob *pJob, SQueryTableRsp *rsp) { if (rsp->tbFName[0]) { SCH_LOCK(SCH_WRITE, &pJob->resLock); - + if (NULL == pJob->execRes.res) { pJob->execRes.res = taosArrayInit(pJob->taskNum, sizeof(STbVerInfo)); if (NULL == pJob->execRes.res) { - SCH_UNLOCK(SCH_WRITE, &pJob->resLock); + SCH_UNLOCK(SCH_WRITE, &pJob->resLock); SCH_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); } } @@ -610,7 +600,6 @@ int32_t schGetTaskInJob(SSchJob *pJob, uint64_t taskId, SSchTask **pTask) { return TSDB_CODE_SUCCESS; } - int32_t schLaunchJob(SSchJob *pJob) { if (EXPLAIN_MODE_STATIC == pJob->attr.explainMode) { SCH_ERR_RET(qExecStaticExplain(pJob->pDag, (SRetrieveTableRsp **)&pJob->fetchRes)); @@ -623,11 +612,10 @@ int32_t schLaunchJob(SSchJob *pJob) { return TSDB_CODE_SUCCESS; } - void schDropJobAllTasks(SSchJob *pJob) { schDropTaskInHashList(pJob, pJob->execTasks); -// schDropTaskInHashList(pJob, pJob->succTasks); -// schDropTaskInHashList(pJob, pJob->failTasks); + // schDropTaskInHashList(pJob, pJob->succTasks); + // schDropTaskInHashList(pJob, pJob->failTasks); } void schFreeJobImpl(void *job) { @@ -659,10 +647,10 @@ void schFreeJobImpl(void *job) { schFreeFlowCtrl(pJob); taosHashCleanup(pJob->execTasks); -// taosHashCleanup(pJob->failTasks); -// taosHashCleanup(pJob->succTasks); + // taosHashCleanup(pJob->failTasks); + // taosHashCleanup(pJob->succTasks); taosHashCleanup(pJob->taskList); - + taosArrayDestroy(pJob->levels); taosArrayDestroy(pJob->nodeList); taosArrayDestroy(pJob->dataSrcTasks); @@ -688,19 +676,19 @@ void schFreeJobImpl(void *job) { } int32_t schJobFetchRows(SSchJob *pJob) { - int32_t code = 0; + int32_t code = 0; if (!(pJob->attr.explainMode == EXPLAIN_MODE_STATIC)) { SCH_ERR_RET(schLaunchFetchTask(pJob)); - + if (schChkCurrentOp(pJob, SCH_OP_FETCH, true)) { SCH_JOB_DLOG("sync wait for rsp now, job status:%s", SCH_GET_JOB_STATUS_STR(pJob)); tsem_wait(&pJob->rspSem); - SCH_RET(schDumpJobFetchRes(pJob, pJob->userRes.fetchRes)); + SCH_RET(schDumpJobFetchRes(pJob, pJob->userRes.fetchRes)); } } else { if (schChkCurrentOp(pJob, SCH_OP_FETCH, true)) { - SCH_RET(schDumpJobFetchRes(pJob, pJob->userRes.fetchRes)); + SCH_RET(schDumpJobFetchRes(pJob, pJob->userRes.fetchRes)); } else { schPostJobRes(pJob, SCH_OP_FETCH); } @@ -736,9 +724,9 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) { } else { pJob->nodeList = taosArrayDup(pReq->pNodeList); } - - pJob->taskList = - taosHashInit(pReq->pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK); + + pJob->taskList = taosHashInit(pReq->pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, + HASH_ENTRY_LOCK); if (NULL == pJob->taskList) { SCH_JOB_ELOG("taosHashInit %d taskList failed", pReq->pDag->numOfSubplans); SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY); @@ -750,8 +738,8 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) { SCH_ERR_JRET(qExecExplainBegin(pReq->pDag, &pJob->explainCtx, pReq->startTs)); } - pJob->execTasks = - taosHashInit(pReq->pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK); + pJob->execTasks = taosHashInit(pReq->pDag->numOfSubplans, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, + HASH_ENTRY_LOCK); if (NULL == pJob->execTasks) { SCH_JOB_ELOG("taosHashInit %d execTasks failed", pReq->pDag->numOfSubplans); SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY); @@ -769,7 +757,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) { *pJobId = pJob->refId; - SCH_JOB_DLOG("job refId:0x%" PRIx64" created", pJob->refId); + SCH_JOB_DLOG("job refId:0x%" PRIx64 " created", pJob->refId); return TSDB_CODE_SUCCESS; @@ -782,31 +770,31 @@ _return: } else { taosRemoveRef(schMgmt.jobRef, pJob->refId); } - + SCH_RET(code); } int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq) { int32_t code = 0; - qDebug("QID:0x%" PRIx64 " sch job refId 0x%"PRIx64 " started", pReq->pDag->queryId, pJob->refId); + qDebug("QID:0x%" PRIx64 " sch job refId 0x%" PRIx64 " started", pReq->pDag->queryId, pJob->refId); SCH_ERR_RET(schLaunchJob(pJob)); - + if (pReq->syncReq) { SCH_JOB_DLOG("sync wait for rsp now, job status:%s", SCH_GET_JOB_STATUS_STR(pJob)); tsem_wait(&pJob->rspSem); } SCH_JOB_DLOG("job exec done, job status:%s, jobId:0x%" PRIx64, SCH_GET_JOB_STATUS_STR(pJob), pJob->refId); - + return TSDB_CODE_SUCCESS; } -void schDirectPostJobRes(SSchedulerReq* pReq, int32_t errCode) { +void schDirectPostJobRes(SSchedulerReq *pReq, int32_t errCode) { if (NULL == pReq || pReq->syncReq) { return; } - + if (pReq->execFp) { (*pReq->execFp)(NULL, pReq->cbParam, errCode); } else if (pReq->fetchFp) { @@ -827,16 +815,17 @@ bool schChkCurrentOp(SSchJob *pJob, int32_t op, int8_t sync) { return r; } -void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int32_t errCode) { +void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq *pReq, int32_t errCode) { int32_t op = 0; - + switch (type) { case SCH_OP_EXEC: if (pReq && pReq->syncReq) { SCH_LOCK(SCH_WRITE, &pJob->opStatus.lock); op = atomic_val_compare_exchange_32(&pJob->opStatus.op, type, SCH_OP_NULL); if (SCH_OP_NULL == op || op != type) { - SCH_JOB_ELOG("job not in %s operation, op:%s, status:%s", schGetOpStr(type), schGetOpStr(op), jobTaskStatusStr(pJob->status)); + SCH_JOB_ELOG("job not in %s operation, op:%s, status:%s", schGetOpStr(type), schGetOpStr(op), + jobTaskStatusStr(pJob->status)); } SCH_UNLOCK(SCH_WRITE, &pJob->opStatus.lock); schDumpJobExecRes(pJob, pReq->pExecRes); @@ -847,7 +836,8 @@ void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int SCH_LOCK(SCH_WRITE, &pJob->opStatus.lock); op = atomic_val_compare_exchange_32(&pJob->opStatus.op, type, SCH_OP_NULL); if (SCH_OP_NULL == op || op != type) { - SCH_JOB_ELOG("job not in %s operation, op:%s, status:%s", schGetOpStr(type), schGetOpStr(op), jobTaskStatusStr(pJob->status)); + SCH_JOB_ELOG("job not in %s operation, op:%s, status:%s", schGetOpStr(type), schGetOpStr(op), + jobTaskStatusStr(pJob->status)); } SCH_UNLOCK(SCH_WRITE, &pJob->opStatus.lock); } @@ -866,10 +856,10 @@ void schProcessOnOpEnd(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int SCH_JOB_DLOG("job end %s operation with code %s", schGetOpStr(type), tstrerror(errCode)); } -int32_t schProcessOnOpBegin(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq) { +int32_t schProcessOnOpBegin(SSchJob *pJob, SCH_OP_TYPE type, SSchedulerReq *pReq) { int32_t code = 0; - int8_t status = SCH_GET_JOB_STATUS(pJob); - + int8_t status = SCH_GET_JOB_STATUS(pJob); + switch (type) { case SCH_OP_EXEC: SCH_LOCK(SCH_WRITE, &pJob->opStatus.lock); @@ -879,9 +869,9 @@ int32_t schProcessOnOpBegin(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq schDirectPostJobRes(pReq, TSDB_CODE_TSC_APP_ERROR); SCH_ERR_RET(TSDB_CODE_TSC_APP_ERROR); } - + SCH_JOB_DLOG("job start %s operation", schGetOpStr(pJob->opStatus.op)); - + pJob->opStatus.syncReq = pReq->syncReq; SCH_UNLOCK(SCH_WRITE, &pJob->opStatus.lock); break; @@ -893,16 +883,16 @@ int32_t schProcessOnOpBegin(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq schDirectPostJobRes(pReq, TSDB_CODE_TSC_APP_ERROR); SCH_ERR_RET(TSDB_CODE_TSC_APP_ERROR); } - + SCH_JOB_DLOG("job start %s operation", schGetOpStr(pJob->opStatus.op)); - + pJob->userRes.fetchRes = pReq->pFetchRes; pJob->userRes.fetchFp = pReq->fetchFp; pJob->userRes.cbParam = pReq->cbParam; - + pJob->opStatus.syncReq = pReq->syncReq; SCH_UNLOCK(SCH_WRITE, &pJob->opStatus.lock); - + if (!SCH_JOB_NEED_FETCH(pJob)) { SCH_JOB_ELOG("no need to fetch data, status:%s", SCH_GET_JOB_STATUS_STR(pJob)); SCH_ERR_RET(TSDB_CODE_QRY_APP_ERROR); @@ -912,7 +902,7 @@ int32_t schProcessOnOpBegin(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq SCH_JOB_ELOG("job status error for fetch, status:%s", jobTaskStatusStr(status)); SCH_ERR_RET(TSDB_CODE_SCH_STATUS_ERROR); } - + break; case SCH_OP_GET_STATUS: if (pJob->status < JOB_TASK_STATUS_INIT || pJob->levelNum <= 0 || NULL == pJob->levels) { @@ -941,23 +931,23 @@ void schProcessOnCbEnd(SSchJob *pJob, SSchTask *pTask, int32_t errCode) { if (errCode) { schHandleJobFailure(pJob, errCode); } - + if (pJob) { schReleaseJob(pJob->refId); } } -int32_t schProcessOnCbBegin(SSchJob** job, SSchTask** task, uint64_t qId, int64_t rId, uint64_t tId) { +int32_t schProcessOnCbBegin(SSchJob **job, SSchTask **task, uint64_t qId, int64_t rId, uint64_t tId) { int32_t code = 0; - int8_t status = 0; + int8_t status = 0; SSchTask *pTask = NULL; - SSchJob *pJob = schAcquireJob(rId); + SSchJob *pJob = schAcquireJob(rId); if (NULL == pJob) { qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, qId, tId, rId); SCH_ERR_RET(TSDB_CODE_QRY_JOB_NOT_EXIST); } - + if (schJobNeedToStop(pJob, &status)) { SCH_TASK_DLOG("will not do further processing cause of job status %s", jobTaskStatusStr(status)); SCH_ERR_JRET(TSDB_CODE_SCH_IGNORE_ERROR); @@ -980,9 +970,6 @@ _return: if (pJob) { schReleaseJob(rId); } - + SCH_RET(code); } - - - diff --git a/source/libs/scheduler/src/schStatus.c b/source/libs/scheduler/src/schStatus.c index 64cda573f1..66d58c95d1 100644 --- a/source/libs/scheduler/src/schStatus.c +++ b/source/libs/scheduler/src/schStatus.c @@ -29,25 +29,25 @@ int32_t schSwitchJobStatus(SSchJob* pJob, int32_t status, void* param) { case JOB_TASK_STATUS_INIT: break; case JOB_TASK_STATUS_EXEC: - SCH_ERR_JRET(schExecJob(pJob, (SSchedulerReq*)param)); + SCH_ERR_JRET(schExecJob(pJob, (SSchedulerReq*)param)); break; case JOB_TASK_STATUS_PART_SUCC: SCH_ERR_JRET(schProcessOnJobPartialSuccess(pJob)); break; case JOB_TASK_STATUS_SUCC: break; - case JOB_TASK_STATUS_FAIL: + case JOB_TASK_STATUS_FAIL: SCH_RET(schProcessOnJobFailure(pJob, (param ? *(int32_t*)param : 0))); break; case JOB_TASK_STATUS_DROP: schProcessOnJobDropped(pJob, *(int32_t*)param); - + if (taosRemoveRef(schMgmt.jobRef, pJob->refId)) { SCH_JOB_ELOG("remove job from job list failed, refId:0x%" PRIx64, pJob->refId); } else { SCH_JOB_DLOG("job removed from jobRef list, refId:0x%" PRIx64, pJob->refId); } - break; + break; default: { SCH_JOB_ELOG("unknown job status %d", status); SCH_RET(TSDB_CODE_SCH_STATUS_ERROR); @@ -62,7 +62,7 @@ _return: } int32_t schHandleOpBeginEvent(int64_t jobId, SSchJob** job, SCH_OP_TYPE type, SSchedulerReq* pReq) { - SSchJob *pJob = schAcquireJob(jobId); + SSchJob* pJob = schAcquireJob(jobId); if (NULL == pJob) { qWarn("Acquire sch job failed, may be dropped, jobId:0x%" PRIx64, jobId); SCH_ERR_RET(TSDB_CODE_SCH_STATUS_ERROR); @@ -75,12 +75,12 @@ int32_t schHandleOpBeginEvent(int64_t jobId, SSchJob** job, SCH_OP_TYPE type, SS int32_t schHandleOpEndEvent(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq, int32_t errCode) { int32_t code = errCode; - + if (NULL == pJob) { schDirectPostJobRes(pReq, errCode); SCH_RET(code); } - + schProcessOnOpEnd(pJob, type, pReq, errCode); if (TSDB_CODE_SCH_IGNORE_ERROR == errCode) { @@ -91,5 +91,3 @@ int32_t schHandleOpEndEvent(SSchJob* pJob, SCH_OP_TYPE type, SSchedulerReq* pReq return code; } - - diff --git a/source/libs/scheduler/src/schTask.c b/source/libs/scheduler/src/schTask.c index b6f96a188e..2678a652a9 100644 --- a/source/libs/scheduler/src/schTask.c +++ b/source/libs/scheduler/src/schTask.c @@ -16,12 +16,12 @@ #include "catalog.h" #include "command.h" #include "query.h" +#include "qworker.h" #include "schInt.h" +#include "tglobal.h" #include "tmsg.h" #include "tref.h" #include "trpc.h" -#include "qworker.h" -#include "tglobal.h" void schFreeTask(SSchJob *pJob, SSchTask *pTask) { schDeregisterTaskHb(pJob, pTask); @@ -94,7 +94,7 @@ int32_t schRecordTaskSucceedNode(SSchJob *pJob, SSchTask *pTask) { if (SCH_IS_LOCAL_EXEC_TASK(pJob, pTask)) { return TSDB_CODE_SUCCESS; } - + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); if (NULL == addr) { SCH_TASK_ELOG("taosArrayGet candidate addr failed, idx:%d, size:%d", pTask->candidateIdx, @@ -162,14 +162,15 @@ 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_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); return TSDB_CODE_SUCCESS; @@ -837,17 +838,18 @@ int32_t schHandleExplainRes(SArray *pExplainRes) { goto _return; } - SSchTask *pTask = NULL; - SSchJob *pJob = NULL; + SSchTask *pTask = NULL; + SSchJob *pJob = NULL; for (int32_t i = 0; i < resNum; ++i) { - SExplainLocalRsp* localRsp = taosArrayGet(pExplainRes, i); + SExplainLocalRsp *localRsp = taosArrayGet(pExplainRes, i); qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ", begin to handle LOCAL explain rsp msg", localRsp->qId, localRsp->tId); pJob = schAcquireJob(localRsp->rId); if (NULL == pJob) { - qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, localRsp->qId, localRsp->tId, localRsp->rId); + qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, localRsp->qId, + localRsp->tId, localRsp->rId); SCH_ERR_JRET(TSDB_CODE_QRY_JOB_NOT_EXIST); } @@ -857,16 +859,17 @@ int32_t schHandleExplainRes(SArray *pExplainRes) { schReleaseJob(pJob->refId); SCH_ERR_JRET(TSDB_CODE_SCH_IGNORE_ERROR); } - + code = schGetTaskInJob(pJob, localRsp->tId, &pTask); if (TSDB_CODE_SUCCESS == code) { code = schProcessExplainRsp(pJob, pTask, &localRsp->rsp); } - + schReleaseJob(pJob->refId); - qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ", end to handle LOCAL explain rsp msg, code:%x", localRsp->qId, localRsp->tId, code); + qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ", end to handle LOCAL explain rsp msg, code:%x", localRsp->qId, + localRsp->tId, code); SCH_ERR_JRET(code); @@ -879,7 +882,7 @@ int32_t schHandleExplainRes(SArray *pExplainRes) { _return: for (int32_t i = 0; i < resNum; ++i) { - SExplainLocalRsp* localRsp = taosArrayGet(pExplainRes, i); + SExplainLocalRsp *localRsp = taosArrayGet(pExplainRes, i); tFreeSExplainRsp(&localRsp->rsp); } @@ -890,7 +893,7 @@ _return: int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask) { SSubplan *plan = pTask->plan; - int32_t code = 0; + int32_t code = 0; if (NULL == pTask->msg) { // TODO add more detailed reason for failure code = qSubPlanToMsg(plan, &pTask->msg, &pTask->msgLen); @@ -899,7 +902,7 @@ int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask) { pTask->msgLen); SCH_ERR_RET(code); } else if (tsQueryPlannerTrace) { - char *msg = NULL; + char *msg = NULL; int32_t msgLen = 0; qSubPlanToString(plan, &msg, &msgLen); SCH_TASK_DLOGL("physical plan len:%d, %s", msgLen, msg); @@ -912,18 +915,18 @@ int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask) { if (SCH_IS_QUERY_JOB(pJob)) { SCH_ERR_RET(schEnsureHbConnection(pJob, pTask)); } - + SCH_RET(schBuildAndSendMsg(pJob, pTask, NULL, plan->msgType)); } int32_t schLaunchLocalTask(SSchJob *pJob, SSchTask *pTask) { - //SCH_ERR_JRET(schSetTaskCandidateAddrs(pJob, pTask)); + // SCH_ERR_JRET(schSetTaskCandidateAddrs(pJob, pTask)); if (NULL == schMgmt.queryMgmt) { SCH_ERR_RET(qWorkerInit(NODE_TYPE_CLIENT, CLIENT_HANDLE, (void **)&schMgmt.queryMgmt, NULL)); } SArray *explainRes = NULL; - SQWMsg qwMsg = {0}; + SQWMsg qwMsg = {0}; qwMsg.msgInfo.taskType = TASK_TYPE_TEMP; qwMsg.msgInfo.explain = SCH_IS_EXPLAIN_JOB(pJob); qwMsg.msgInfo.needFetch = SCH_TASK_NEED_FETCH(pTask); @@ -934,8 +937,9 @@ int32_t schLaunchLocalTask(SSchJob *pJob, SSchTask *pTask) { if (SCH_IS_EXPLAIN_JOB(pJob)) { explainRes = taosArrayInit(pJob->taskNum, sizeof(SExplainLocalRsp)); } - - SCH_ERR_RET(qWorkerProcessLocalQuery(schMgmt.queryMgmt, schMgmt.sId, pJob->queryId, pTask->taskId, pJob->refId, pTask->execId, &qwMsg, explainRes)); + + SCH_ERR_RET(qWorkerProcessLocalQuery(schMgmt.queryMgmt, schMgmt.sId, pJob->queryId, pTask->taskId, pJob->refId, + pTask->execId, &qwMsg, explainRes)); if (SCH_IS_EXPLAIN_JOB(pJob)) { SCH_ERR_RET(schHandleExplainRes(explainRes)); @@ -958,17 +962,17 @@ int32_t schLaunchTaskImpl(void *param) { if (pCtx->asyncLaunch) { SCH_LOCK_TASK(pTask); } - - int8_t status = 0; - int32_t code = 0; + + int8_t status = 0; + int32_t code = 0; atomic_add_fetch_32(&pTask->level->taskLaunchedNum, 1); pTask->execId++; pTask->retryTimes++; pTask->waitRetry = false; - SCH_TASK_DLOG("start to launch %s task, execId %d, retry %d", SCH_IS_LOCAL_EXEC_TASK(pJob, pTask) ? "LOCAL" : "REMOTE", - pTask->execId, pTask->retryTimes); + SCH_TASK_DLOG("start to launch %s task, execId %d, retry %d", + SCH_IS_LOCAL_EXEC_TASK(pJob, pTask) ? "LOCAL" : "REMOTE", pTask->execId, pTask->retryTimes); SCH_LOG_TASK_START_TS(pTask); @@ -1086,19 +1090,20 @@ int32_t schExecRemoteFetch(SSchJob *pJob, SSchTask *pTask) { } int32_t schExecLocalFetch(SSchJob *pJob, SSchTask *pTask) { - void *pRsp = NULL; + void *pRsp = NULL; SArray *explainRes = NULL; if (SCH_IS_EXPLAIN_JOB(pJob)) { explainRes = taosArrayInit(pJob->taskNum, sizeof(SExplainLocalRsp)); } - SCH_ERR_RET(qWorkerProcessLocalFetch(schMgmt.queryMgmt, schMgmt.sId, pJob->queryId, pTask->taskId, pJob->refId, pTask->execId, &pRsp, explainRes)); + SCH_ERR_RET(qWorkerProcessLocalFetch(schMgmt.queryMgmt, schMgmt.sId, pJob->queryId, pTask->taskId, pJob->refId, + pTask->execId, &pRsp, explainRes)); if (SCH_IS_EXPLAIN_JOB(pJob)) { SCH_ERR_RET(schHandleExplainRes(explainRes)); } - + SCH_ERR_RET(schProcessFetchRsp(pJob, pTask, pRsp, TSDB_CODE_SUCCESS)); return TSDB_CODE_SUCCESS; diff --git a/source/libs/scheduler/src/schUtil.c b/source/libs/scheduler/src/schUtil.c index 6f12780ff9..7099dbb2c0 100644 --- a/source/libs/scheduler/src/schUtil.c +++ b/source/libs/scheduler/src/schUtil.c @@ -21,21 +21,21 @@ #include "tref.h" #include "trpc.h" -FORCE_INLINE SSchJob *schAcquireJob(int64_t refId) { - qDebug("sch acquire jobId:0x%"PRIx64, refId); - return (SSchJob *)taosAcquireRef(schMgmt.jobRef, refId); +FORCE_INLINE SSchJob *schAcquireJob(int64_t refId) { + qDebug("sch acquire jobId:0x%" PRIx64, refId); + return (SSchJob *)taosAcquireRef(schMgmt.jobRef, refId); } -FORCE_INLINE int32_t schReleaseJob(int64_t refId) { +FORCE_INLINE int32_t schReleaseJob(int64_t refId) { if (0 == refId) { return TSDB_CODE_SUCCESS; } - - qDebug("sch release jobId:0x%"PRIx64, refId); - return taosReleaseRef(schMgmt.jobRef, refId); + + qDebug("sch release jobId:0x%" PRIx64, refId); + return taosReleaseRef(schMgmt.jobRef, refId); } -char* schGetOpStr(SCH_OP_TYPE type) { +char *schGetOpStr(SCH_OP_TYPE type) { switch (type) { case SCH_OP_NULL: return "NULL"; @@ -53,28 +53,28 @@ char* schGetOpStr(SCH_OP_TYPE type) { void schFreeHbTrans(SSchHbTrans *pTrans) { rpcReleaseHandle(pTrans->trans.pHandle, TAOS_CONN_CLIENT); - schFreeRpcCtx(&pTrans->rpcCtx); + schFreeRpcCtx(&pTrans->rpcCtx); } -void schCleanClusterHb(void* pTrans) { +void schCleanClusterHb(void *pTrans) { SCH_LOCK(SCH_WRITE, &schMgmt.hbLock); SSchHbTrans *hb = taosHashIterate(schMgmt.hbConnections, NULL); while (hb) { if (hb->trans.pTrans == pTrans) { - SQueryNodeEpId* pEpId = taosHashGetKey(hb, NULL); + SQueryNodeEpId *pEpId = taosHashGetKey(hb, NULL); schFreeHbTrans(hb); taosHashRemove(schMgmt.hbConnections, pEpId, sizeof(SQueryNodeEpId)); } - + hb = taosHashIterate(schMgmt.hbConnections, hb); } - + SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock); } int32_t schRemoveHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *epId) { - int32_t code = 0; + int32_t code = 0; SCH_LOCK(SCH_WRITE, &schMgmt.hbLock); SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId)); @@ -94,7 +94,6 @@ int32_t schRemoveHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *ep return TSDB_CODE_SUCCESS; } - int32_t schAddHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *epId, bool *exist) { int32_t code = 0; SSchHbTrans hb = {0}; @@ -155,13 +154,13 @@ void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask) { if (!pTask->registerdHb) { return; } - + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); SQueryNodeEpId epId = {0}; epId.nodeId = addr->nodeId; - SEp* pEp = SCH_GET_CUR_EP(addr); + SEp *pEp = SCH_GET_CUR_EP(addr); strcpy(epId.ep.fqdn, pEp->fqdn); epId.ep.port = pEp->port; @@ -180,24 +179,22 @@ void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask) { } else { SCH_UNLOCK(SCH_READ, &schMgmt.hbLock); } - + pTask->registerdHb = false; } - - int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) { SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); SQueryNodeEpId epId = {0}; epId.nodeId = addr->nodeId; - SEp* pEp = SCH_GET_CUR_EP(addr); + SEp *pEp = SCH_GET_CUR_EP(addr); strcpy(epId.ep.fqdn, pEp->fqdn); epId.ep.port = pEp->port; SCH_ERR_RET(schRegisterHbConnection(pJob, pTask, &epId)); - + pTask->registerdHb = true; return TSDB_CODE_SUCCESS; @@ -226,7 +223,6 @@ int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans) { return TSDB_CODE_SUCCESS; } - void schCloseJobRef(void) { if (!atomic_load_8((int8_t *)&schMgmt.exit)) { return; @@ -242,7 +238,7 @@ uint64_t schGenTaskId(void) { return atomic_add_fetch_64(&schMgmt.taskId, 1); } uint64_t schGenUUID(void) { static uint64_t hashId = 0; - static int32_t requestSerialId = 0; + static int32_t requestSerialId = 0; if (hashId == 0) { char uid[64] = {0}; @@ -254,15 +250,14 @@ uint64_t schGenUUID(void) { } } - int64_t ts = taosGetTimestampMs(); - uint64_t pid = taosGetPId(); - int32_t val = atomic_add_fetch_32(&requestSerialId, 1); + int64_t ts = taosGetTimestampMs(); + uint64_t pid = taosGetPId(); + int32_t val = atomic_add_fetch_32(&requestSerialId, 1); uint64_t id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF); return id; } - void schFreeRpcCtxVal(const void *arg) { if (NULL == arg) { return; @@ -307,5 +302,3 @@ int32_t schGetTaskFromList(SHashObj *pTaskList, uint64_t taskId, SSchTask **pTas return TSDB_CODE_SUCCESS; } - - diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c index 0ccaef3857..813a11c056 100644 --- a/source/libs/scheduler/src/scheduler.c +++ b/source/libs/scheduler/src/scheduler.c @@ -14,10 +14,10 @@ */ #include "query.h" +#include "qworker.h" #include "schInt.h" #include "tmsg.h" #include "tref.h" -#include "qworker.h" SSchedulerMgmt schMgmt = { .jobRef = -1, @@ -35,7 +35,7 @@ int32_t schedulerInit() { schMgmt.cfg.enableReSchedule = true; qDebug("schedule policy init to %d", schMgmt.cfg.schPolicy); - + schMgmt.jobRef = taosOpenRef(schMgmt.cfg.maxJobNum, schFreeJobImpl); if (schMgmt.jobRef < 0) { qError("init schduler jobRef failed, num:%u", schMgmt.cfg.maxJobNum); @@ -61,7 +61,7 @@ int32_t schedulerInit() { int32_t schedulerExecJob(SSchedulerReq *pReq, int64_t *pJobId) { qDebug("scheduler %s exec job start", pReq->syncReq ? "SYNC" : "ASYNC"); - int32_t code = 0; + int32_t code = 0; SSchJob *pJob = NULL; SCH_ERR_JRET(schInitJob(pJobId, pReq)); @@ -73,7 +73,7 @@ int32_t schedulerExecJob(SSchedulerReq *pReq, int64_t *pJobId) { SCH_ERR_JRET(schSwitchJobStatus(pJob, JOB_TASK_STATUS_EXEC, pReq)); _return: - + SCH_RET(schHandleOpEndEvent(pJob, SCH_OP_EXEC, pReq, code)); } @@ -144,7 +144,7 @@ int32_t schedulerEnableReSchedule(bool enableResche) { return TSDB_CODE_SUCCESS; } -void schedulerFreeJob(int64_t* jobId, int32_t errCode) { +void schedulerFreeJob(int64_t *jobId, int32_t errCode) { if (0 == *jobId) { return; } @@ -158,7 +158,7 @@ void schedulerFreeJob(int64_t* jobId, int32_t errCode) { SCH_JOB_DLOG("start to free job 0x%" PRIx64 ", errCode:0x%x", *jobId, errCode); schHandleJobDrop(pJob, errCode); - + schReleaseJob(*jobId); *jobId = 0; } diff --git a/source/libs/scheduler/test/schedulerTests.cpp b/source/libs/scheduler/test/schedulerTests.cpp index ca2122ed8f..97e14b617c 100644 --- a/source/libs/scheduler/test/schedulerTests.cpp +++ b/source/libs/scheduler/test/schedulerTests.cpp @@ -30,15 +30,12 @@ #endif #include "os.h" -#include "tglobal.h" -#include "taos.h" -#include "tdef.h" -#include "tvariant.h" #include "catalog.h" #include "scheduler.h" #include "taos.h" #include "tdatablock.h" #include "tdef.h" +#include "tglobal.h" #include "trpc.h" #include "tvariant.h" @@ -56,8 +53,9 @@ namespace { -extern "C" int32_t schHandleResponseMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *msg, int32_t msgSize, int32_t rspCode); -extern "C" int32_t schHandleCallback(void* param, const SDataBuf* pMsg, int32_t msgType, int32_t rspCode); +extern "C" int32_t schHandleResponseMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *msg, int32_t msgSize, + int32_t rspCode); +extern "C" int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode); int64_t insertJobRefId = 0; int64_t queryJobRefId = 0; @@ -66,16 +64,15 @@ uint64_t schtMergeTemplateId = 0x4; uint64_t schtFetchTaskId = 0; uint64_t schtQueryId = 1; -bool schtTestStop = false; -bool schtTestDeadLoop = false; +bool schtTestStop = false; +bool schtTestDeadLoop = false; int32_t schtTestMTRunSec = 10; int32_t schtTestPrintNum = 1000; int32_t schtStartFetch = 0; - void schtInitLogFile() { - const char *defaultLogFileNamePrefix = "taoslog"; - const int32_t maxLogFileNum = 10; + const char *defaultLogFileNamePrefix = "taoslog"; + const int32_t maxLogFileNum = 10; tsAsyncLog = 0; qDebugFlag = 159; @@ -84,24 +81,22 @@ void schtInitLogFile() { if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) { printf("failed to open log file in directory:%s\n", tsLogDir); } - } -void schtQueryCb(SExecResult* pResult, void* param, int32_t code) { +void schtQueryCb(SExecResult *pResult, void *param, int32_t code) { assert(TSDB_CODE_SUCCESS == code); - *(int32_t*)param = 1; + *(int32_t *)param = 1; } - void schtBuildQueryDag(SQueryPlan *dag) { uint64_t qId = schtQueryId; - + dag->queryId = qId; dag->numOfSubplans = 2; dag->pSubplans = nodesMakeList(); - SNodeListNode *scan = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST); - SNodeListNode *merge = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST); - + SNodeListNode *scan = (SNodeListNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); + SNodeListNode *merge = (SNodeListNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); + SSubplan *scanPlan = (SSubplan *)taosMemoryCalloc(1, sizeof(SSubplan)); SSubplan *mergePlan = (SSubplan *)taosMemoryCalloc(1, sizeof(SSubplan)); @@ -117,7 +112,7 @@ void schtBuildQueryDag(SQueryPlan *dag) { scanPlan->pChildren = NULL; scanPlan->level = 1; scanPlan->pParents = nodesMakeList(); - scanPlan->pNode = (SPhysiNode*)taosMemoryCalloc(1, sizeof(SPhysiNode)); + scanPlan->pNode = (SPhysiNode *)taosMemoryCalloc(1, sizeof(SPhysiNode)); scanPlan->msgType = TDMT_SCH_QUERY; mergePlan->id.queryId = qId; @@ -129,32 +124,32 @@ void schtBuildQueryDag(SQueryPlan *dag) { mergePlan->pChildren = nodesMakeList(); mergePlan->pParents = NULL; - mergePlan->pNode = (SPhysiNode*)taosMemoryCalloc(1, sizeof(SPhysiNode)); + mergePlan->pNode = (SPhysiNode *)taosMemoryCalloc(1, sizeof(SPhysiNode)); mergePlan->msgType = TDMT_SCH_QUERY; merge->pNodeList = nodesMakeList(); scan->pNodeList = nodesMakeList(); - nodesListAppend(merge->pNodeList, (SNode*)mergePlan); - nodesListAppend(scan->pNodeList, (SNode*)scanPlan); + nodesListAppend(merge->pNodeList, (SNode *)mergePlan); + nodesListAppend(scan->pNodeList, (SNode *)scanPlan); - nodesListAppend(mergePlan->pChildren, (SNode*)scanPlan); - nodesListAppend(scanPlan->pParents, (SNode*)mergePlan); + nodesListAppend(mergePlan->pChildren, (SNode *)scanPlan); + nodesListAppend(scanPlan->pParents, (SNode *)mergePlan); - nodesListAppend(dag->pSubplans, (SNode*)merge); - nodesListAppend(dag->pSubplans, (SNode*)scan); + nodesListAppend(dag->pSubplans, (SNode *)merge); + nodesListAppend(dag->pSubplans, (SNode *)scan); } void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) { uint64_t qId = schtQueryId; - int32_t scanPlanNum = 20; - + int32_t scanPlanNum = 20; + dag->queryId = qId; dag->numOfSubplans = 2; dag->pSubplans = nodesMakeList(); - SNodeListNode *scan = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST); - SNodeListNode *merge = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST); - + SNodeListNode *scan = (SNodeListNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); + SNodeListNode *merge = (SNodeListNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); + SSubplan *scanPlan = (SSubplan *)taosMemoryCalloc(scanPlanNum, sizeof(SSubplan)); SSubplan *mergePlan = (SSubplan *)taosMemoryCalloc(1, sizeof(SSubplan)); @@ -180,13 +175,13 @@ void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) { scanPlan[i].pChildren = NULL; scanPlan[i].level = 1; scanPlan[i].pParents = nodesMakeList(); - scanPlan[i].pNode = (SPhysiNode*)taosMemoryCalloc(1, sizeof(SPhysiNode)); + scanPlan[i].pNode = (SPhysiNode *)taosMemoryCalloc(1, sizeof(SPhysiNode)); scanPlan[i].msgType = TDMT_SCH_QUERY; - nodesListAppend(scanPlan[i].pParents, (SNode*)mergePlan); - nodesListAppend(mergePlan->pChildren, (SNode*)(scanPlan + i)); + nodesListAppend(scanPlan[i].pParents, (SNode *)mergePlan); + nodesListAppend(mergePlan->pChildren, (SNode *)(scanPlan + i)); - nodesListAppend(scan->pNodeList, (SNode*)(scanPlan + i)); + nodesListAppend(scan->pNodeList, (SNode *)(scanPlan + i)); } mergePlan->id.queryId = qId; @@ -197,29 +192,25 @@ void schtBuildQueryFlowCtrlDag(SQueryPlan *dag) { mergePlan->execNode.epSet.numOfEps = 0; mergePlan->pParents = NULL; - mergePlan->pNode = (SPhysiNode*)taosMemoryCalloc(1, sizeof(SPhysiNode)); + mergePlan->pNode = (SPhysiNode *)taosMemoryCalloc(1, sizeof(SPhysiNode)); mergePlan->msgType = TDMT_SCH_QUERY; - nodesListAppend(merge->pNodeList, (SNode*)mergePlan); - - nodesListAppend(dag->pSubplans, (SNode*)merge); - nodesListAppend(dag->pSubplans, (SNode*)scan); -} - - -void schtFreeQueryDag(SQueryPlan *dag) { + nodesListAppend(merge->pNodeList, (SNode *)mergePlan); + nodesListAppend(dag->pSubplans, (SNode *)merge); + nodesListAppend(dag->pSubplans, (SNode *)scan); } +void schtFreeQueryDag(SQueryPlan *dag) {} void schtBuildInsertDag(SQueryPlan *dag) { uint64_t qId = 0x0000000000000002; - + dag->queryId = qId; dag->numOfSubplans = 2; dag->pSubplans = nodesMakeList(); - SNodeListNode *inserta = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST); - + SNodeListNode *inserta = (SNodeListNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); + SSubplan *insertPlan = (SSubplan *)taosMemoryCalloc(2, sizeof(SSubplan)); insertPlan[0].id.queryId = qId; @@ -235,7 +226,7 @@ void schtBuildInsertDag(SQueryPlan *dag) { insertPlan[0].pChildren = NULL; insertPlan[0].pParents = NULL; insertPlan[0].pNode = NULL; - insertPlan[0].pDataSink = (SDataSinkNode*)taosMemoryCalloc(1, sizeof(SDataSinkNode)); + insertPlan[0].pDataSink = (SDataSinkNode *)taosMemoryCalloc(1, sizeof(SDataSinkNode)); insertPlan[0].msgType = TDMT_VND_SUBMIT; insertPlan[1].id.queryId = qId; @@ -251,48 +242,43 @@ void schtBuildInsertDag(SQueryPlan *dag) { insertPlan[1].pChildren = NULL; insertPlan[1].pParents = NULL; insertPlan[1].pNode = NULL; - insertPlan[1].pDataSink = (SDataSinkNode*)taosMemoryCalloc(1, sizeof(SDataSinkNode)); + insertPlan[1].pDataSink = (SDataSinkNode *)taosMemoryCalloc(1, sizeof(SDataSinkNode)); insertPlan[1].msgType = TDMT_VND_SUBMIT; inserta->pNodeList = nodesMakeList(); - nodesListAppend(inserta->pNodeList, (SNode*)insertPlan); + nodesListAppend(inserta->pNodeList, (SNode *)insertPlan); insertPlan += 1; - nodesListAppend(inserta->pNodeList, (SNode*)insertPlan); + nodesListAppend(inserta->pNodeList, (SNode *)insertPlan); - nodesListAppend(dag->pSubplans, (SNode*)inserta); + nodesListAppend(dag->pSubplans, (SNode *)inserta); } - -int32_t schtPlanToString(const SSubplan *subplan, char** str, int32_t* len) { +int32_t schtPlanToString(const SSubplan *subplan, char **str, int32_t *len) { *str = (char *)taosMemoryCalloc(1, 20); *len = 20; return 0; } -void schtExecNode(SSubplan* subplan, uint64_t groupId, SQueryNodeAddr* ep) { +void schtExecNode(SSubplan *subplan, uint64_t groupId, SQueryNodeAddr *ep) {} -} - -void schtRpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *pRid) { - -} +void schtRpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *pRid) {} void schtSetPlanToString() { static Stub stub; stub.set(qSubPlanToString, schtPlanToString); { #ifdef WINDOWS - AddrAny any; - std::map result; + AddrAny any; + std::map result; any.get_func_addr("qSubPlanToString", result); #endif #ifdef LINUX - AddrAny any("libplanner.so"); - std::map result; + AddrAny any("libplanner.so"); + std::map result; any.get_global_func_addr_dynsym("^qSubPlanToString$", result); #endif - for (const auto& f : result) { + for (const auto &f : result) { stub.set(f.second, schtPlanToString); } } @@ -303,16 +289,16 @@ void schtSetExecNode() { stub.set(qSetSubplanExecutionNode, schtExecNode); { #ifdef WINDOWS - AddrAny any; - std::map result; + AddrAny any; + std::map result; any.get_func_addr("qSetSubplanExecutionNode", result); #endif #ifdef LINUX - AddrAny any("libplanner.so"); - std::map result; + AddrAny any("libplanner.so"); + std::map result; any.get_global_func_addr_dynsym("^qSetSubplanExecutionNode$", result); #endif - for (const auto& f : result) { + for (const auto &f : result) { stub.set(f.second, schtExecNode); } } @@ -323,22 +309,22 @@ void schtSetRpcSendRequest() { stub.set(rpcSendRequest, schtRpcSendRequest); { #ifdef WINDOWS - AddrAny any; - std::map result; + AddrAny any; + std::map result; any.get_func_addr("rpcSendRequest", result); #endif #ifdef LINUX - AddrAny any("libtransport.so"); - std::map result; + AddrAny any("libtransport.so"); + std::map result; any.get_global_func_addr_dynsym("^rpcSendRequest$", result); #endif - for (const auto& f : result) { + for (const auto &f : result) { stub.set(f.second, schtRpcSendRequest); } } } -int32_t schtAsyncSendMsgToServer(void *pTransporter, SEpSet* epSet, int64_t* pTransporterId, SMsgSendInfo* pInfo) { +int32_t schtAsyncSendMsgToServer(void *pTransporter, SEpSet *epSet, int64_t *pTransporterId, SMsgSendInfo *pInfo) { if (pInfo) { taosMemoryFreeClear(pInfo->param); taosMemoryFreeClear(pInfo->msgInfo.pData); @@ -347,32 +333,30 @@ int32_t schtAsyncSendMsgToServer(void *pTransporter, SEpSet* epSet, int64_t* pTr return 0; } - void schtSetAsyncSendMsgToServer() { static Stub stub; stub.set(asyncSendMsgToServer, schtAsyncSendMsgToServer); { #ifdef WINDOWS - AddrAny any; - std::map result; + AddrAny any; + std::map result; any.get_func_addr("asyncSendMsgToServer", result); #endif #ifdef LINUX - AddrAny any("libtransport.so"); - std::map result; + AddrAny any("libtransport.so"); + std::map result; any.get_global_func_addr_dynsym("^asyncSendMsgToServer$", result); #endif - for (const auto& f : result) { + for (const auto &f : result) { stub.set(f.second, schtAsyncSendMsgToServer); } } } - void *schtSendRsp(void *param) { SSchJob *pJob = NULL; - int64_t job = 0; - int32_t code = 0; + int64_t job = 0; + int32_t code = 0; while (true) { job = *(int64_t *)param; @@ -384,7 +368,7 @@ void *schtSendRsp(void *param) { } pJob = schAcquireJob(job); - + void *pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { SSchTask *task = *(SSchTask **)pIter; @@ -392,9 +376,9 @@ void *schtSendRsp(void *param) { SSubmitRsp rsp = {0}; rsp.affectedRows = 10; schHandleResponseMsg(pJob, task, TDMT_VND_SUBMIT_RSP, (char *)&rsp, sizeof(rsp), 0); - + pIter = taosHashIterate(pJob->execTasks, pIter); - } + } schReleaseJob(job); @@ -402,28 +386,27 @@ void *schtSendRsp(void *param) { } void *schtCreateFetchRspThread(void *param) { - int64_t job = *(int64_t *)param; - SSchJob* pJob = schAcquireJob(job); + int64_t job = *(int64_t *)param; + SSchJob *pJob = schAcquireJob(job); taosSsleep(1); - int32_t code = 0; + int32_t code = 0; SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)taosMemoryCalloc(1, sizeof(SRetrieveTableRsp)); rsp->completed = 1; rsp->numOfRows = 10; - + code = schHandleResponseMsg(pJob, pJob->fetchTask, TDMT_SCH_FETCH_RSP, (char *)rsp, sizeof(*rsp), 0); schReleaseJob(job); - + assert(code == 0); return NULL; } - void *schtFetchRspThread(void *aa) { - SDataBuf dataBuf = {0}; - SSchTaskCallbackParam* param = NULL; + SDataBuf dataBuf = {0}; + SSchTaskCallbackParam *param = NULL; while (!schtTestStop) { if (0 == atomic_val_compare_exchange_32(&schtStartFetch, 1, 0)) { @@ -431,13 +414,13 @@ void *schtFetchRspThread(void *aa) { } taosUsleep(1); - + param = (SSchTaskCallbackParam *)taosMemoryCalloc(1, sizeof(*param)); - param->queryId = schtQueryId; + param->queryId = schtQueryId; param->taskId = schtFetchTaskId; - int32_t code = 0; + int32_t code = 0; SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)taosMemoryCalloc(1, sizeof(SRetrieveTableRsp)); rsp->completed = 1; rsp->numOfRows = 10; @@ -446,7 +429,7 @@ void *schtFetchRspThread(void *aa) { dataBuf.len = sizeof(*rsp); code = schHandleCallback(param, &dataBuf, TDMT_SCH_FETCH_RSP, 0); - + assert(code == 0 || code); } return NULL; @@ -454,8 +437,8 @@ void *schtFetchRspThread(void *aa) { void schtFreeQueryJob(int32_t freeThread) { static uint32_t freeNum = 0; - int64_t job = queryJobRefId; - + int64_t job = queryJobRefId; + if (job && atomic_val_compare_exchange_64(&queryJobRefId, job, 0)) { schedulerFreeJob(&job, 0); if (freeThread) { @@ -466,31 +449,29 @@ void schtFreeQueryJob(int32_t freeThread) { } } -void* schtRunJobThread(void *aa) { - void *mockPointer = (void *)0x1; - char *clusterId = "cluster1"; - char *dbname = "1.db1"; - char *tablename = "table1"; +void *schtRunJobThread(void *aa) { + void *mockPointer = (void *)0x1; + char *clusterId = "cluster1"; + char *dbname = "1.db1"; + char *tablename = "table1"; SVgroupInfo vgInfo = {0}; - SQueryPlan dag; + SQueryPlan dag; schtInitLogFile(); - int32_t code = schedulerInit(); assert(code == 0); - schtSetPlanToString(); schtSetExecNode(); schtSetAsyncSendMsgToServer(); - SSchJob *pJob = NULL; + SSchJob *pJob = NULL; SSchTaskCallbackParam *param = NULL; - SHashObj *execTasks = NULL; - SDataBuf dataBuf = {0}; - uint32_t jobFinished = 0; - int32_t queryDone = 0; + SHashObj *execTasks = NULL; + SDataBuf dataBuf = {0}; + uint32_t jobFinished = 0; + int32_t queryDone = 0; while (!schtTestStop) { schtBuildQueryDag(&dag); @@ -503,10 +484,10 @@ void* schtRunJobThread(void *aa) { taosArrayPush(qnodeList, &qnodeAddr); queryDone = 0; - + SRequestConnInfo conn = {0}; conn.pTrans = mockPointer; - SSchedulerReq req = {0}; + SSchedulerReq req = {0}; req.syncReq = false; req.pConn = &conn; req.pNodeList = qnodeList; @@ -514,8 +495,8 @@ void* schtRunJobThread(void *aa) { req.sql = "select * from tb"; req.execFp = schtQueryCb; req.cbParam = &queryDone; - - code = schedulerExecJob(&req, &queryJobRefId); + + code = schedulerExecJob(&req, &queryJobRefId); assert(code == 0); pJob = schAcquireJob(queryJobRefId); @@ -524,20 +505,20 @@ void* schtRunJobThread(void *aa) { schtFreeQueryDag(&dag); continue; } - + execTasks = taosHashInit(5, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK); void *pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { SSchTask *task = *(SSchTask **)pIter; schtFetchTaskId = task->taskId - 1; - + taosHashPut(execTasks, &task->taskId, sizeof(task->taskId), task, sizeof(*task)); pIter = taosHashIterate(pJob->execTasks, pIter); - } + } param = (SSchTaskCallbackParam *)taosMemoryCalloc(1, sizeof(*param)); param->refId = queryJobRefId; - param->queryId = pJob->queryId; + param->queryId = pJob->queryId; pIter = taosHashIterate(execTasks, NULL); while (pIter) { @@ -547,16 +528,16 @@ void* schtRunJobThread(void *aa) { SQueryTableRsp rsp = {0}; dataBuf.pData = &rsp; dataBuf.len = sizeof(rsp); - + code = schHandleCallback(param, &dataBuf, TDMT_SCH_QUERY_RSP, 0); assert(code == 0 || code); pIter = taosHashIterate(execTasks, pIter); - } + } param = (SSchTaskCallbackParam *)taosMemoryCalloc(1, sizeof(*param)); param->refId = queryJobRefId; - param->queryId = pJob->queryId; + param->queryId = pJob->queryId; pIter = taosHashIterate(execTasks, NULL); while (pIter) { @@ -566,28 +547,27 @@ void* schtRunJobThread(void *aa) { SQueryTableRsp rsp = {0}; dataBuf.pData = &rsp; dataBuf.len = sizeof(rsp); - + code = schHandleCallback(param, &dataBuf, TDMT_SCH_QUERY_RSP, 0); assert(code == 0 || code); - - pIter = taosHashIterate(execTasks, pIter); - } + pIter = taosHashIterate(execTasks, pIter); + } while (true) { if (queryDone) { break; } - + taosUsleep(10000); } atomic_store_32(&schtStartFetch, 1); - void *data = NULL; + void *data = NULL; req.syncReq = true; req.pFetchRes = &data; - + code = schedulerFetchRows(queryJobRefId, &req); assert(code == 0 || code); @@ -600,7 +580,7 @@ void* schtRunJobThread(void *aa) { data = NULL; code = schedulerFetchRows(queryJobRefId, &req); assert(code == 0 || code); - + schtFreeQueryJob(0); taosHashCleanup(execTasks); @@ -620,7 +600,7 @@ void* schtRunJobThread(void *aa) { return NULL; } -void* schtFreeJobThread(void *aa) { +void *schtFreeJobThread(void *aa) { while (!schtTestStop) { taosUsleep(taosRand() % 100); schtFreeQueryJob(1); @@ -628,17 +608,16 @@ void* schtFreeJobThread(void *aa) { return NULL; } - -} +} // namespace TEST(queryTest, normalCase) { - void *mockPointer = (void *)0x1; - char *clusterId = "cluster1"; - char *dbname = "1.db1"; - char *tablename = "table1"; + void *mockPointer = (void *)0x1; + char *clusterId = "cluster1"; + char *dbname = "1.db1"; + char *tablename = "table1"; SVgroupInfo vgInfo = {0}; - int64_t job = 0; - SQueryPlan dag; + int64_t job = 0; + SQueryPlan dag; memset(&dag, 0, sizeof(dag)); @@ -648,7 +627,7 @@ TEST(queryTest, normalCase) { strcpy(qnodeAddr.fqdn, "qnode0.ep"); qnodeAddr.port = 6031; taosArrayPush(qnodeList, &qnodeAddr); - + int32_t code = schedulerInit(); ASSERT_EQ(code, 0); @@ -662,30 +641,29 @@ TEST(queryTest, normalCase) { SRequestConnInfo conn = {0}; conn.pTrans = mockPointer; - SSchedulerReq req = {0}; + SSchedulerReq req = {0}; req.pConn = &conn; req.pNodeList = qnodeList; req.pDag = &dag; req.sql = "select * from tb"; req.execFp = schtQueryCb; req.cbParam = &queryDone; - - code = schedulerExecJob(&req, &job); + + code = schedulerExecJob(&req, &job); ASSERT_EQ(code, 0); - SSchJob *pJob = schAcquireJob(job); - + void *pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { SSchTask *task = *(SSchTask **)pIter; SQueryTableRsp rsp = {0}; code = schHandleResponseMsg(pJob, task, TDMT_SCH_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0); - + ASSERT_EQ(code, 0); pIter = taosHashIterate(pJob->execTasks, pIter); - } + } pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { @@ -693,10 +671,10 @@ TEST(queryTest, normalCase) { SQueryTableRsp rsp = {0}; code = schHandleResponseMsg(pJob, task, TDMT_SCH_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0); - + ASSERT_EQ(code, 0); pIter = taosHashIterate(pJob->execTasks, pIter); - } + } while (true) { if (queryDone) { @@ -705,14 +683,14 @@ TEST(queryTest, normalCase) { taosUsleep(10000); } - + TdThreadAttr thattr; taosThreadAttrInit(&thattr); TdThread thread1; taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job); - void *data = NULL; + void *data = NULL; req.syncReq = true; req.pFetchRes = &data; @@ -739,13 +717,13 @@ TEST(queryTest, normalCase) { } TEST(queryTest, readyFirstCase) { - void *mockPointer = (void *)0x1; - char *clusterId = "cluster1"; - char *dbname = "1.db1"; - char *tablename = "table1"; + void *mockPointer = (void *)0x1; + char *clusterId = "cluster1"; + char *dbname = "1.db1"; + char *tablename = "table1"; SVgroupInfo vgInfo = {0}; - int64_t job = 0; - SQueryPlan dag; + int64_t job = 0; + SQueryPlan dag; memset(&dag, 0, sizeof(dag)); @@ -755,7 +733,7 @@ TEST(queryTest, readyFirstCase) { strcpy(qnodeAddr.fqdn, "qnode0.ep"); qnodeAddr.port = 6031; taosArrayPush(qnodeList, &qnodeAddr); - + int32_t code = schedulerInit(); ASSERT_EQ(code, 0); @@ -765,11 +743,11 @@ TEST(queryTest, readyFirstCase) { schtSetExecNode(); schtSetAsyncSendMsgToServer(); - int32_t queryDone = 0; + int32_t queryDone = 0; SRequestConnInfo conn = {0}; conn.pTrans = mockPointer; - SSchedulerReq req = {0}; + SSchedulerReq req = {0}; req.pConn = &conn; req.pNodeList = qnodeList; req.pDag = &dag; @@ -779,19 +757,18 @@ TEST(queryTest, readyFirstCase) { code = schedulerExecJob(&req, &job); ASSERT_EQ(code, 0); - SSchJob *pJob = schAcquireJob(job); - + void *pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { SSchTask *task = *(SSchTask **)pIter; SQueryTableRsp rsp = {0}; code = schHandleResponseMsg(pJob, task, TDMT_SCH_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0); - + ASSERT_EQ(code, 0); pIter = taosHashIterate(pJob->execTasks, pIter); - } + } pIter = taosHashIterate(pJob->execTasks, NULL); while (pIter) { @@ -799,10 +776,10 @@ TEST(queryTest, readyFirstCase) { SQueryTableRsp rsp = {0}; code = schHandleResponseMsg(pJob, task, TDMT_SCH_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0); - + ASSERT_EQ(code, 0); pIter = taosHashIterate(pJob->execTasks, pIter); - } + } while (true) { if (queryDone) { @@ -812,14 +789,13 @@ TEST(queryTest, readyFirstCase) { taosUsleep(10000); } - TdThreadAttr thattr; taosThreadAttrInit(&thattr); TdThread thread1; taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job); - void *data = NULL; + void *data = NULL; req.syncReq = true; req.pFetchRes = &data; code = schedulerFetchRows(job, &req); @@ -844,28 +820,26 @@ TEST(queryTest, readyFirstCase) { schedulerDestroy(); } - - TEST(queryTest, flowCtrlCase) { - void *mockPointer = (void *)0x1; - char *clusterId = "cluster1"; - char *dbname = "1.db1"; - char *tablename = "table1"; + void *mockPointer = (void *)0x1; + char *clusterId = "cluster1"; + char *dbname = "1.db1"; + char *tablename = "table1"; SVgroupInfo vgInfo = {0}; - int64_t job = 0; - SQueryPlan dag; + int64_t job = 0; + SQueryPlan dag; schtInitLogFile(); taosSeedRand(taosGetTimestampSec()); - + SArray *qnodeList = taosArrayInit(1, sizeof(SEp)); SEp qnodeAddr = {0}; strcpy(qnodeAddr.fqdn, "qnode0.ep"); qnodeAddr.port = 6031; taosArrayPush(qnodeList, &qnodeAddr); - + int32_t code = schedulerInit(); ASSERT_EQ(code, 0); @@ -875,10 +849,10 @@ TEST(queryTest, flowCtrlCase) { schtSetExecNode(); schtSetAsyncSendMsgToServer(); - int32_t queryDone = 0; + int32_t queryDone = 0; SRequestConnInfo conn = {0}; conn.pTrans = mockPointer; - SSchedulerReq req = {0}; + SSchedulerReq req = {0}; req.pConn = &conn; req.pNodeList = qnodeList; req.pDag = &dag; @@ -889,17 +863,16 @@ TEST(queryTest, flowCtrlCase) { code = schedulerExecJob(&req, &job); ASSERT_EQ(code, 0); - SSchJob *pJob = schAcquireJob(job); bool qDone = false; - + while (!qDone) { void *pIter = taosHashIterate(pJob->execTasks, NULL); if (NULL == pIter) { break; } - + while (pIter) { SSchTask *task = *(SSchTask **)pIter; @@ -908,15 +881,15 @@ TEST(queryTest, flowCtrlCase) { if (task->lastMsgType == TDMT_SCH_QUERY) { SQueryTableRsp rsp = {0}; code = schHandleResponseMsg(pJob, task, TDMT_SCH_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0); - + ASSERT_EQ(code, 0); } else { qDone = true; break; } - + pIter = NULL; - } + } } while (true) { @@ -933,7 +906,7 @@ TEST(queryTest, flowCtrlCase) { TdThread thread1; taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job); - void *data = NULL; + void *data = NULL; req.syncReq = true; req.pFetchRes = &data; code = schedulerFetchRows(job, &req); @@ -958,15 +931,14 @@ TEST(queryTest, flowCtrlCase) { schedulerDestroy(); } - TEST(insertTest, normalCase) { - void *mockPointer = (void *)0x1; - char *clusterId = "cluster1"; - char *dbname = "1.db1"; - char *tablename = "table1"; + void *mockPointer = (void *)0x1; + char *clusterId = "cluster1"; + char *dbname = "1.db1"; + char *tablename = "table1"; SVgroupInfo vgInfo = {0}; - SQueryPlan dag; - uint64_t numOfRows = 0; + SQueryPlan dag; + uint64_t numOfRows = 0; SArray *qnodeList = taosArrayInit(1, sizeof(SEp)); @@ -974,7 +946,7 @@ TEST(insertTest, normalCase) { strcpy(qnodeAddr.fqdn, "qnode0.ep"); qnodeAddr.port = 6031; taosArrayPush(qnodeList, &qnodeAddr); - + int32_t code = schedulerInit(); ASSERT_EQ(code, 0); @@ -993,21 +965,21 @@ TEST(insertTest, normalCase) { SRequestConnInfo conn = {0}; conn.pTrans = mockPointer; - SSchedulerReq req = {0}; + SSchedulerReq req = {0}; req.pConn = &conn; req.pNodeList = qnodeList; req.pDag = &dag; req.sql = "insert into tb values(now,1)"; req.execFp = schtQueryCb; req.cbParam = NULL; - + code = schedulerExecJob(&req, &insertJobRefId); ASSERT_EQ(code, 0); ASSERT_EQ(res.numOfRows, 20); schedulerFreeJob(&insertJobRefId, 0); - schedulerDestroy(); + schedulerDestroy(); } TEST(multiThread, forceFree) { @@ -1027,12 +999,12 @@ TEST(multiThread, forceFree) { break; } } - + schtTestStop = true; taosSsleep(3); } -int main(int argc, char** argv) { +int main(int argc, char **argv) { taosSeedRand(taosGetTimestampSec()); testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/tools/scripts/codeFormat.sh b/tools/scripts/codeFormat.sh index 8707dcdabc..5ececa980e 100644 --- a/tools/scripts/codeFormat.sh +++ b/tools/scripts/codeFormat.sh @@ -12,25 +12,7 @@ FORMAT_DIR_LIST=( "${PRJ_ROOT_DIR}/source/os" "${PRJ_ROOT_DIR}/source/util" "${PRJ_ROOT_DIR}/source/common" - "${PRJ_ROOT_DIR}/source/libs/cache" - "${PRJ_ROOT_DIR}/source/libs/catalog" - "${PRJ_ROOT_DIR}/source/libs/command" - "${PRJ_ROOT_DIR}/source/libs/executor" - "${PRJ_ROOT_DIR}/source/libs/function" - "${PRJ_ROOT_DIR}/source/libs/index" - "${PRJ_ROOT_DIR}/source/libs/monitor" - "${PRJ_ROOT_DIR}/source/libs/nodes" - # "${PRJ_ROOT_DIR}/source/libs/parser" - "${PRJ_ROOT_DIR}/source/libs/planner" - "${PRJ_ROOT_DIR}/source/libs/qcom" - "${PRJ_ROOT_DIR}/source/libs/qworker" - "${PRJ_ROOT_DIR}/source/libs/scalar" - "${PRJ_ROOT_DIR}/source/libs/stream" - "${PRJ_ROOT_DIR}/source/libs/sync" - "${PRJ_ROOT_DIR}/source/libs/tdb" - "${PRJ_ROOT_DIR}/source/libs/tfs" - "${PRJ_ROOT_DIR}/source/libs/transport" - "${PRJ_ROOT_DIR}/source/libs/wal" + "${PRJ_ROOT_DIR}/source/libs" "${PRJ_ROOT_DIR}/source/client/inc" "${PRJ_ROOT_DIR}/source/client/src" "${PRJ_ROOT_DIR}/source/client/test" @@ -45,7 +27,7 @@ EXCLUDE_FILE_LIST=( ) for d in ${FORMAT_DIR_LIST[@]}; do - for f in $(find $d -type f -regex '.*\.\(cpp\|hpp\|c\|h\)'); do + for f in $(find $d -type f -not -name "*/sql.c" -regex '.*\.\(cpp\|hpp\|c\|h\)'); do ${FORMAT_BIN} -i $f done done From a014be5ca95b6b6d63321754533afc8219c3ee25 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Thu, 13 Oct 2022 17:06:04 +0800 Subject: [PATCH 2/2] more code --- tools/scripts/codeFormat.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) mode change 100644 => 100755 tools/scripts/codeFormat.sh diff --git a/tools/scripts/codeFormat.sh b/tools/scripts/codeFormat.sh old mode 100644 new mode 100755 index 5ececa980e..cc7db5c723 --- a/tools/scripts/codeFormat.sh +++ b/tools/scripts/codeFormat.sh @@ -27,7 +27,7 @@ EXCLUDE_FILE_LIST=( ) for d in ${FORMAT_DIR_LIST[@]}; do - for f in $(find $d -type f -not -name "*/sql.c" -regex '.*\.\(cpp\|hpp\|c\|h\)'); do + for f in $(find $d -type f -not -name '*sql.c' -regex '.*\.\(cpp\|hpp\|c\|h\)'); do ${FORMAT_BIN} -i $f done done