diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 0ded2bc4b5..ed0f5cd31d 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -92,6 +92,10 @@ extern int32_t tsQueryNodeChunkSize; extern bool tsQueryUseNodeAllocator; extern bool tsKeepColumnName; extern bool tsEnableQueryHb; +extern int32_t tsRedirectPeriod; +extern int32_t tsRedirectFactor; +extern int32_t tsRedirectMaxPeriod; +extern int32_t tsMaxRetryWaitTime; // client extern int32_t tsMinSlidingTime; diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h index 96ccef7164..f51aa88485 100644 --- a/include/libs/qcom/query.h +++ b/include/libs/qcom/query.h @@ -259,9 +259,15 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t #define NEED_CLIENT_HANDLE_ERROR(_code) \ (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \ NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code)) + +#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR) +#define SYNC_SELF_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR) +#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) (false) // used later + #define NEED_REDIRECT_ERROR(_code) \ ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || \ - (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || (_code) == TSDB_CODE_SYN_NOT_LEADER || \ + (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \ + SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || \ (_code) == TSDB_CODE_APP_NOT_READY || (_code) == TSDB_CODE_RPC_BROKEN_LINK) #define NEED_CLIENT_RM_TBLMETA_REQ(_type) \ @@ -270,7 +276,8 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t #define NEED_SCHEDULER_REDIRECT_ERROR(_code) \ ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || \ - (_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_APP_NOT_READY) + SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || \ + SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || (_code) == TSDB_CODE_APP_NOT_READY) #define REQUEST_TOTAL_EXEC_TIMES 2 diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h index 8cc37910fd..d761813db1 100644 --- a/include/libs/transport/trpc.h +++ b/include/libs/transport/trpc.h @@ -85,6 +85,11 @@ typedef struct SRpcInit { int32_t retryLimit; // retry limit int32_t retryInterval; // retry interval ms + int32_t retryMinInterval; // retry init interval + int32_t retryStepFactor; // retry interval factor + int32_t retryMaxInterval; // retry max interval + int64_t retryMaxTimouet; + int32_t compressSize; // -1: no compress, 0 : all data compressed, size: compress data if larger than size int8_t encryption; // encrypt or not diff --git a/include/util/taoserror.h b/include/util/taoserror.h index e5d0bcb249..25d37020cc 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -92,6 +92,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_NO_AVAIL_DISK TAOS_DEF_ERROR_CODE(0, 0x0129) #define TSDB_CODE_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x012A) #define TSDB_CODE_NO_DISKSPACE TAOS_DEF_ERROR_CODE(0, 0x012B) +#define TSDB_CODE_TIMEOUT_ERROR TAOS_DEF_ERROR_CODE(0, 0x012C) //client #define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200) @@ -415,6 +416,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_SYN_PROPOSE_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x0911) #define TSDB_CODE_SYN_STANDBY_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x0912) #define TSDB_CODE_SYN_BATCH_ERROR TAOS_DEF_ERROR_CODE(0, 0x0913) +#define TSDB_CODE_SYN_RESTORING TAOS_DEF_ERROR_CODE(0, 0x0914) #define TSDB_CODE_SYN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x09FF) // tq diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index 3f4e1bb513..6f1414b72d 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -148,6 +148,10 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) { rpcInit.dfp = destroyAhandle; rpcInit.retryLimit = tsRpcRetryLimit; rpcInit.retryInterval = tsRpcRetryInterval; + rpcInit.retryMinInterval = tsRedirectPeriod; + rpcInit.retryStepFactor = tsRedirectFactor; + rpcInit.retryMaxInterval = tsRedirectMaxPeriod; + rpcInit.retryMaxTimouet = tsMaxRetryWaitTime; void *pDnodeConn = rpcOpen(&rpcInit); if (pDnodeConn == NULL) { diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index aeef1b5277..b6c039f375 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -16,9 +16,9 @@ #define _DEFAULT_SOURCE #include "tglobal.h" #include "tconfig.h" -#include "tmisce.h" #include "tgrant.h" #include "tlog.h" +#include "tmisce.h" GRANT_CFG_DECLARE; @@ -86,6 +86,10 @@ bool tsQueryPlannerTrace = false; int32_t tsQueryNodeChunkSize = 32 * 1024; bool tsQueryUseNodeAllocator = true; bool tsKeepColumnName = false; +int32_t tsRedirectPeriod = 10; +int32_t tsRedirectFactor = 2; +int32_t tsRedirectMaxPeriod = 1000; +int32_t tsMaxRetryWaitTime = 10000; /* * denote if the server needs to compress response message at the application layer to client, including query rsp, @@ -120,7 +124,7 @@ int32_t tsMinIntervalTime = 1; int32_t tsMaxMemUsedByInsert = 1024; float tsSelectivityRatio = 1.0; -int32_t tsTagFilterResCacheSize = 1024*10; +int32_t tsTagFilterResCacheSize = 1024 * 10; // the maximum allowed query buffer size during query processing for each data node. // -1 no limit (default) @@ -305,6 +309,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1; if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1; if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1; + if (cfgAddInt32(pCfg, "maxRetryWaitTime", tsMaxRetryWaitTime, 0, 86400000, 0) != 0) return -1; tsNumOfTaskQueueThreads = tsNumOfCores / 2; tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4); @@ -659,6 +664,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) { tsRpcRetryLimit = cfgGetItem(pCfg, "rpcRetryLimit")->i32; tsRpcRetryInterval = cfgGetItem(pCfg, "rpcRetryInterval")->i32; + tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32; return 0; } @@ -874,6 +880,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32; } else if (strcasecmp("maxMemUsedByInsert", name) == 0) { tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32; + } else if (strcasecmp("maxRetryWaitTime", name) == 0) { + tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32; } break; } diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index f78fd33e47..95656fd76c 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -258,8 +258,13 @@ int32_t dmInitClient(SDnode *pDnode) { rpcInit.parent = pDnode; rpcInit.rfp = rpcRfp; rpcInit.compressSize = tsCompressMsgSize; + rpcInit.retryLimit = tsRpcRetryLimit; rpcInit.retryInterval = tsRpcRetryInterval; + rpcInit.retryMinInterval = tsRedirectPeriod; + rpcInit.retryStepFactor = tsRedirectFactor; + rpcInit.retryMaxInterval = tsRedirectMaxPeriod; + rpcInit.retryMaxTimouet = tsMaxRetryWaitTime; pTrans->clientRpc = rpcOpen(&rpcInit); if (pTrans->clientRpc == NULL) { diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index 4324c412f7..6c205b5522 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -66,7 +66,7 @@ void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg) { } pMsg->info.hasEpSet = 1; - SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info, .msgType = pMsg->msgType + 1}; + SRpcMsg rsp = {.code = TSDB_CODE_SYN_NOT_LEADER, .info = pMsg->info, .msgType = pMsg->msgType + 1}; tmsgSendRedirectRsp(&rsp, &newEpSet); } diff --git a/source/libs/scheduler/inc/schInt.h b/source/libs/scheduler/inc/schInt.h index 6884824ba9..dfc48e7d9f 100644 --- a/source/libs/scheduler/inc/schInt.h +++ b/source/libs/scheduler/inc/schInt.h @@ -27,6 +27,7 @@ extern "C" { #include "tarray.h" #include "thash.h" #include "trpc.h" +#include "ttimer.h" enum { SCH_READ = 1, @@ -146,6 +147,7 @@ typedef struct SSchedulerMgmt { int32_t jobRef; int32_t jobNum; SSchStat stat; + void *timer; SRWLatch hbLock; SHashObj *hbConnections; void *queryMgmt; @@ -202,12 +204,30 @@ typedef struct SSchTaskProfile { int64_t endTs; } SSchTaskProfile; +typedef struct SSchRedirectCtx { + int32_t periodMs; + bool inRedirect; + int32_t totalTimes; + int32_t roundTotal; + int32_t roundTimes; // retry times in current round + int64_t startTs; +} SSchRedirectCtx; + +typedef struct SSchTimerParam { + int64_t rId; + uint64_t queryId; + uint64_t taskId; +} SSchTimerParam; + 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 + int32_t delayExecMs; // task execution delay time + tmr_h delayTimer; // task delay execution timer + SSchRedirectCtx redirectCtx; // task redirect context bool waitRetry; // wait for retry int32_t execId; // task current execute index SSchLevel *level; // level @@ -488,6 +508,7 @@ extern SSchedulerMgmt schMgmt; void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask); void schCleanClusterHb(void *pTrans); int32_t schLaunchTask(SSchJob *job, SSchTask *task); +int32_t schDelayLaunchTask(SSchJob *pJob, SSchTask *pTask); int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType); SSchJob *schAcquireJob(int64_t refId); int32_t schReleaseJob(int64_t refId); @@ -529,6 +550,7 @@ 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 *schDumpEpSet(SEpSet *pEpSet); 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); diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index c154060d21..1a6d7df349 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -887,9 +887,14 @@ int32_t schAsyncSendMsg(SSchJob *pJob, SSchTask *pTask, SSchTrans *trans, SQuery SCH_ERR_JRET(schGenerateCallBackInfo(pJob, pTask, msg, msgSize, msgType, trans, isHb, &pMsgSendInfo)); SCH_ERR_JRET(schUpdateSendTargetInfo(pMsgSendInfo, addr, pTask)); - qDebug("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId, - epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle); - + if (pJob && pTask) { + SCH_TASK_DLOG("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId, + epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle); + } else { + qDebug("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId, + epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle); + } + if (pTask) { pTask->lastMsgType = msgType; } diff --git a/source/libs/scheduler/src/schTask.c b/source/libs/scheduler/src/schTask.c index 7e5b3faedb..9a7f3332b3 100644 --- a/source/libs/scheduler/src/schTask.c +++ b/source/libs/scheduler/src/schTask.c @@ -340,6 +340,70 @@ int32_t schRescheduleTask(SSchJob *pJob, SSchTask *pTask) { return TSDB_CODE_SUCCESS; } +int32_t schChkUpdateRedirectCtx(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet) { + SSchRedirectCtx *pCtx = &pTask->redirectCtx; + if (!pCtx->inRedirect) { + pCtx->inRedirect = true; + pCtx->periodMs = tsRedirectPeriod; + pCtx->startTs = taosGetTimestampMs(); + + if (SCH_IS_DATA_BIND_TASK(pTask)) { + if (pEpSet) { + pCtx->roundTotal = pEpSet->numOfEps; + } else { + SQueryNodeAddr *pAddr = taosArrayGet(pTask->candidateAddrs, 0); + pCtx->roundTotal = pAddr->epSet.numOfEps; + } + } else { + pCtx->roundTotal = 1; + } + + goto _return; + } + + pCtx->totalTimes++; + + if (SCH_IS_DATA_BIND_TASK(pTask) && pEpSet) { + pCtx->roundTotal = pEpSet->numOfEps; + pCtx->roundTimes = 0; + + pTask->delayExecMs = 0; + + goto _return; + } + + pCtx->roundTimes++; + + if (pCtx->roundTimes >= pCtx->roundTotal) { + int64_t nowTs = taosGetTimestampMs(); + int64_t lastTime = nowTs - pCtx->startTs; + if (lastTime > tsMaxRetryWaitTime) { + SCH_TASK_DLOG("task no more redirect retry since timeout, now:%" PRId64 ", start:%" PRId64 ", max:%d, total:%d", + nowTs, pCtx->startTs, tsMaxRetryWaitTime, pCtx->totalTimes); + SCH_ERR_RET(TSDB_CODE_TIMEOUT_ERROR); + } + + pCtx->periodMs *= tsRedirectFactor; + if (pCtx->periodMs > tsRedirectMaxPeriod) { + pCtx->periodMs = tsRedirectMaxPeriod; + } + + int64_t leftTime = tsMaxRetryWaitTime - lastTime; + pTask->delayExecMs = leftTime < pCtx->periodMs ? leftTime : pCtx->periodMs; + + goto _return; + } + + pTask->delayExecMs = 0; + +_return: + + SCH_TASK_DLOG("task start %d/%d/%d redirect retry, delayExec:%d", pCtx->roundTimes, pCtx->roundTotal, + pCtx->totalTimes, pTask->delayExecMs); + + return TSDB_CODE_SUCCESS; +} + int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32_t rspCode) { int32_t code = 0; @@ -349,14 +413,10 @@ int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32 pTask->retryTimes = 0; } - if (((pTask->execId + 1) >= pTask->maxExecTimes) || ((pTask->retryTimes + 1) > pTask->maxRetryTimes)) { - SCH_TASK_DLOG("task no more retry since reach max times %d:%d, execId %d", pTask->maxRetryTimes, - pTask->maxExecTimes, pTask->execId); - schHandleJobFailure(pJob, rspCode); - return TSDB_CODE_SUCCESS; - } + SCH_ERR_JRET(schChkUpdateRedirectCtx(pJob, pTask, pData ? pData->pEpSet : NULL)); pTask->waitRetry = true; + schDropTaskOnExecNode(pJob, pTask); taosHashClear(pTask->execNodes); schRemoveTaskFromExecList(pJob, pTask); @@ -368,8 +428,17 @@ int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32 memset(&pTask->succeedAddr, 0, sizeof(pTask->succeedAddr)); if (SCH_IS_DATA_BIND_TASK(pTask)) { - if (pData) { + if (pData && pData->pEpSet) { SCH_ERR_JRET(schUpdateTaskCandidateAddr(pJob, pTask, pData->pEpSet)); + } else if (SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(rspCode)) { + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); + SCH_SWITCH_EPSET(addr); + SCH_TASK_DLOG("switch task target node %d epset to %d/%d", addr->nodeId, addr->epSet.inUse, addr->epSet.numOfEps); + } else { + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); + SEp *pEp = &addr->epSet.eps[addr->epSet.inUse]; + SCH_TASK_DLOG("task retry node %d current ep, idx:%d/%d,%s:%d", addr->nodeId, addr->epSet.inUse, + addr->epSet.numOfEps, pEp->fqdn, pEp->port); } if (SCH_TASK_NEED_FLOW_CTRL(pJob, pTask)) { @@ -380,7 +449,7 @@ int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32 SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_INIT); - SCH_ERR_JRET(schLaunchTask(pJob, pTask)); + SCH_ERR_JRET(schDelayLaunchTask(pJob, pTask)); return TSDB_CODE_SUCCESS; } @@ -428,28 +497,24 @@ int32_t schHandleRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32 schUpdateJobStatus(pJob, JOB_TASK_STATUS_EXEC); } - if (SCH_IS_DATA_BIND_TASK(pTask)) { + if (SYNC_OTHER_LEADER_REDIRECT_ERROR(rspCode)) { if (NULL == pData->pEpSet) { - SCH_TASK_ELOG("no epset updated while got error %s", tstrerror(rspCode)); - code = rspCode; + SCH_TASK_ELOG("epset updating excepted, error:%s", tstrerror(rspCode)); + code = TSDB_CODE_INVALID_MSG; goto _return; } } code = schDoTaskRedirect(pJob, pTask, pData, rspCode); - taosMemoryFree(pData->pData); - taosMemoryFree(pData->pEpSet); - pData->pData = NULL; - pData->pEpSet = NULL; + taosMemoryFreeClear(pData->pData); + taosMemoryFreeClear(pData->pEpSet); SCH_RET(code); _return: - taosMemoryFree(pData->pData); - taosMemoryFree(pData->pEpSet); - pData->pData = NULL; - pData->pEpSet = NULL; + taosMemoryFreeClear(pData->pData); + taosMemoryFreeClear(pData->pEpSet); SCH_RET(schProcessOnTaskFailure(pJob, pTask, code)); } @@ -715,10 +780,13 @@ int32_t schUpdateTaskCandidateAddr(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSe SQueryNodeAddr *pAddr = taosArrayGet(pTask->candidateAddrs, 0); - SEp *pOld = &pAddr->epSet.eps[pAddr->epSet.inUse]; - SEp *pNew = &pEpSet->eps[pEpSet->inUse]; + char *origEpset = schDumpEpSet(&pAddr->epSet); + char *newEpset = schDumpEpSet(pEpSet); - SCH_TASK_DLOG("update task ep from %s:%d to %s:%d", pOld->fqdn, pOld->port, pNew->fqdn, pNew->port); + SCH_TASK_DLOG("update task target node %d epset from %s to %s", pAddr->nodeId, origEpset, newEpset); + + taosMemoryFree(origEpset); + taosMemoryFree(newEpset); memcpy(&pAddr->epSet, pEpSet, sizeof(pAddr->epSet)); @@ -1078,6 +1146,56 @@ _return: SCH_RET(schProcessOnTaskFailure(pJob, pTask, code)); } +void schHandleTimerEvent(void *param, void *tmrId) { + SSchTimerParam *pTimerParam = (SSchTimerParam *)param; + SSchTask *pTask = NULL; + SSchJob *pJob = NULL; + int32_t code = 0; + + int64_t rId = pTimerParam->rId; + uint64_t queryId = pTimerParam->queryId; + uint64_t taskId = pTimerParam->taskId; + taosMemoryFree(pTimerParam); + + if (schProcessOnCbBegin(&pJob, &pTask, queryId, rId, taskId)) { + return; + } + + code = schLaunchTask(pJob, pTask); + + schProcessOnCbEnd(pJob, pTask, code); +} + +int32_t schDelayLaunchTask(SSchJob *pJob, SSchTask *pTask) { + if (pTask->delayExecMs > 0) { + SSchTimerParam *param = taosMemoryMalloc(sizeof(SSchTimerParam)); + if (NULL == param) { + SCH_TASK_ELOG("taosMemoryMalloc %d failed", (int)sizeof(SSchTimerParam)); + SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); + } + + param->rId = pJob->refId; + param->queryId = pJob->queryId; + param->taskId = pTask->taskId; + + if (NULL == pTask->delayTimer) { + pTask->delayTimer = taosTmrStart(schHandleTimerEvent, pTask->delayExecMs, (void *)param, schMgmt.timer); + if (NULL == pTask->delayTimer) { + SCH_TASK_ELOG("start delay timer failed, handle:%p", schMgmt.timer); + SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); + } + + return TSDB_CODE_SUCCESS; + } + + taosTmrReset(schHandleTimerEvent, pTask->delayExecMs, (void *)param, schMgmt.timer, &pTask->delayTimer); + + return TSDB_CODE_SUCCESS; + } + + SCH_RET(schLaunchTask(pJob, pTask)); +} + int32_t schLaunchLevelTasks(SSchJob *pJob, SSchLevel *level) { SCH_ERR_RET(schChkJobNeedFlowCtrl(pJob, level)); @@ -1099,7 +1217,12 @@ void schDropTaskInHashList(SSchJob *pJob, SHashObj *list) { while (pIter) { SSchTask *pTask = *(SSchTask **)pIter; + SCH_LOCK_TASK(pTask); + if (pTask->delayTimer) { + taosTmrStopA(&pTask->delayTimer); + } schDropTaskOnExecNode(pJob, pTask); + SCH_UNLOCK_TASK(pTask); pIter = taosHashIterate(list, pIter); } diff --git a/source/libs/scheduler/src/schUtil.c b/source/libs/scheduler/src/schUtil.c index 1f1288fcfd..a5a15cd1c6 100644 --- a/source/libs/scheduler/src/schUtil.c +++ b/source/libs/scheduler/src/schUtil.c @@ -36,6 +36,27 @@ FORCE_INLINE int32_t schReleaseJob(int64_t refId) { return taosReleaseRef(schMgmt.jobRef, refId); } +char *schDumpEpSet(SEpSet *pEpSet) { + if (NULL == pEpSet) { + return NULL; + } + + int32_t maxSize = 1024; + char *str = taosMemoryMalloc(maxSize); + if (NULL == str) { + return NULL; + } + + int32_t n = 0; + n += snprintf(str + n, maxSize - n, "numOfEps:%d, inUse:%d eps:", pEpSet->numOfEps, pEpSet->inUse); + for (int32_t i = 0; i < pEpSet->numOfEps; ++i) { + SEp *pEp = &pEpSet->eps[i]; + n += snprintf(str + n, maxSize - n, "[%s:%d]", pEp->fqdn, pEp->port); + } + + return str; +} + char *schGetOpStr(SCH_OP_TYPE type) { switch (type) { case SCH_OP_NULL: diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c index c4b45649a4..91c3b5d7a1 100644 --- a/source/libs/scheduler/src/scheduler.c +++ b/source/libs/scheduler/src/scheduler.c @@ -48,6 +48,12 @@ int32_t schedulerInit() { SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); } + schMgmt.timer = taosTmrInit(0, 0, 0, "scheduler"); + if (NULL == schMgmt.timer) { + qError("init timer failed, error:%s", tstrerror(terrno)); + SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); + } + if (taosGetSystemUUID((char *)&schMgmt.sId, sizeof(schMgmt.sId))) { qError("generate schdulerId failed, errno:%d", errno); SCH_ERR_RET(TSDB_CODE_QRY_SYS_ERROR); diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index ac54749ae1..9add91cdeb 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -137,15 +137,23 @@ typedef struct { tmsg_t msgType; // message type int8_t connType; // connection type cli/srv - int8_t retryCnt; - int8_t retryLimit; - STransCtx appCtx; // STransMsg* pRsp; // for synchronous API tsem_t* pSem; // for synchronous API SCvtAddr cvtAddr; bool setMaxRetry; + int32_t retryMinInterval; + int32_t retryMaxInterval; + int32_t retryStepFactor; + int64_t retryMaxTimeout; + int64_t retryInitTimestamp; + int64_t retryNextInterval; + bool retryInit; + int32_t retryStep; + + int8_t epsetRetryCnt; + int hThrdIdx; } STransConnCtx; diff --git a/source/libs/transport/inc/transportInt.h b/source/libs/transport/inc/transportInt.h index c8a56081cc..833937aa41 100644 --- a/source/libs/transport/inc/transportInt.h +++ b/source/libs/transport/inc/transportInt.h @@ -52,6 +52,11 @@ typedef struct { int32_t retryLimit; // retry limit int32_t retryInterval; // retry interval ms + int32_t retryMinInterval; // retry init interval + int32_t retryStepFactor; // retry interval factor + int32_t retryMaxInterval; // retry max interval + int32_t retryMaxTimouet; + void (*cfp)(void* parent, SRpcMsg*, SEpSet*); bool (*retry)(int32_t code, tmsg_t msgType); bool (*startTimer)(int32_t code, tmsg_t msgType); diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index 88888f2f84..c6a5cfdc95 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -51,6 +51,11 @@ void* rpcOpen(const SRpcInit* pInit) { pRpc->retryLimit = pInit->retryLimit; pRpc->retryInterval = pInit->retryInterval; + pRpc->retryMinInterval = pInit->retryMinInterval; // retry init interval + pRpc->retryStepFactor = pInit->retryStepFactor; + pRpc->retryMaxInterval = pInit->retryMaxInterval; + pRpc->retryMaxTimouet = pInit->retryMaxTimouet; + // register callback handle pRpc->cfp = pInit->cfp; pRpc->retry = pInit->rfp; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 71cc14493f..180540ba6b 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -234,7 +234,7 @@ static void cliWalkCb(uv_handle_t* handle, void* arg); #define REQUEST_PERSIS_HANDLE(msg) ((msg)->info.persistHandle == 1) #define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release) -#define EPSET_IS_VALID(epSet) ((epSet) != NULL && (epSet)->numOfEps != 0) +#define EPSET_IS_VALID(epSet) ((epSet) != NULL && (epSet)->numOfEps >= 0 && (epSet)->inUse >= 0) #define EPSET_GET_SIZE(epSet) (epSet)->numOfEps #define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn) #define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port) @@ -367,7 +367,7 @@ void cliHandleResp(SCliConn* conn) { STraceId* trace = &transMsg.info.traceId; tGDebug("%s conn %p %s received from %s, local info:%s, len:%d, code str:%s", CONN_GET_INST_LABEL(conn), conn, - TMSG_INFO(pHead->msgType), conn->dst, conn->src, msgLen, tstrerror(transMsg.code)); + TMSG_INFO(pHead->msgType), conn->dst, conn->src, pHead->msgLen, tstrerror(transMsg.code)); if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) { tDebug("%s except, conn %p read while cli ignore it", CONN_GET_INST_LABEL(conn), conn); @@ -971,7 +971,7 @@ FORCE_INLINE void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) { FORCE_INLINE bool cliIsEpsetUpdated(int32_t code, STransConnCtx* pCtx) { if (code != 0) return false; - if (pCtx->retryCnt == 0) return false; + // if (pCtx->retryCnt == 0) return false; if (transEpSetIsEqual(&pCtx->epSet, &pCtx->origEpSet)) return false; return true; } @@ -1008,9 +1008,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) { STransConnCtx* pCtx = pMsg->ctx; cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr); + + char tbuf[256] = {0}; + EPSET_DEBUG_STR(&pCtx->epSet, tbuf); + tDebug("current epset %s", tbuf); + if (!EPSET_IS_VALID(&pCtx->epSet)) { - destroyCmsg(pMsg); tError("invalid epset"); + destroyCmsg(pMsg); return; } @@ -1047,12 +1052,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) { addr.sin_family = AF_INET; addr.sin_addr.s_addr = cliGetIpFromFqdnCache(pThrd->fqdn2ipCache, conn->ip); addr.sin_port = (uint16_t)htons((uint16_t)conn->port); - tTrace("%s conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port); + + STraceId* trace = &(pMsg->msg.info.traceId); + tGTrace("%s conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port); int ret = uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, cliConnCb); if (ret != 0) { - tTrace("%s conn %p failed to connect to %s:%d, reason:%s", pTransInst->label, conn, conn->ip, conn->port, - uv_err_name(ret)); + tGTrace("%s conn %p failed to connect to %s:%d, reason:%s", pTransInst->label, conn, conn->ip, conn->port, + uv_err_name(ret)); uv_timer_stop(conn->timer); conn->timer->data = NULL; @@ -1378,13 +1385,14 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) { STraceId* trace = &pMsg->msg.info.traceId; char tbuf[256] = {0}; EPSET_DEBUG_STR(&pCtx->epSet, tbuf); - tGDebug("%s retry on next node, use %s, retryCnt:%d, limit:%d", transLabel(pThrd->pTransInst), tbuf, - pCtx->retryCnt + 1, pCtx->retryLimit); + tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", transLabel(pThrd->pTransInst), tbuf, + pCtx->retryStep, pCtx->retryNextInterval); STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); arg->param1 = pMsg; arg->param2 = pThrd; - transDQSched(pThrd->delayQueue, doDelayTask, arg, pTransInst->retryInterval); + + transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval); } FORCE_INLINE void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) { @@ -1418,6 +1426,122 @@ FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) { *dst = epset; return true; } +bool cliResetEpset(STransConnCtx* pCtx, STransMsg* pResp, bool hasEpSet) { + bool noDelay = true; + if (hasEpSet == false) { + // assert(pResp->contLen == 0); + if (pResp->contLen == 0) { + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + EPSET_FORWARD_INUSE(&pCtx->epSet); + } + } else { + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + EPSET_FORWARD_INUSE(&pCtx->epSet); + } + } + } else { + SEpSet epSet; + int32_t valid = tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet); + if (valid < 0) { + tDebug("get invalid epset, epset equal, continue"); + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + EPSET_FORWARD_INUSE(&pCtx->epSet); + } + } else { + if (!transEpSetIsEqual(&pCtx->epSet, &epSet)) { + tDebug("epset not equal, retry new epset"); + pCtx->epSet = epSet; + noDelay = false; + } else { + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + tDebug("epset equal, continue"); + EPSET_FORWARD_INUSE(&pCtx->epSet); + } + } + } + } + return noDelay; +} +bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { + SCliThrd* pThrd = pConn->hostThrd; + STrans* pTransInst = pThrd->pTransInst; + + STransConnCtx* pCtx = pMsg->ctx; + int32_t code = pResp->code; + + bool retry = pTransInst->retry != NULL ? pTransInst->retry(code, pResp->msgType - 1) : false; + if (retry == false) { + return false; + } + + if (!pCtx->retryInit) { + pCtx->retryMinInterval = pTransInst->retryMinInterval; + pCtx->retryMaxInterval = pTransInst->retryMaxInterval; + pCtx->retryStepFactor = pTransInst->retryStepFactor; + pCtx->retryMaxTimeout = pTransInst->retryMaxTimouet; + pCtx->retryInitTimestamp = taosGetTimestampMs(); + pCtx->retryNextInterval = pCtx->retryMinInterval; + pCtx->retryStep = 0; + pCtx->retryInit = true; + } + if (-1 != pCtx->retryMaxTimeout && taosGetTimestampMs() - pCtx->retryInitTimestamp >= pCtx->retryMaxTimeout) { + return false; + } + + bool noDelay = false; + if (code == TSDB_CODE_RPC_BROKEN_LINK || code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { + tDebug("code str %s, contlen:%d 0", tstrerror(code), pResp->contLen); + noDelay = cliResetEpset(pCtx, pResp, false); + transFreeMsg(pResp->pCont); + transUnrefCliHandle(pConn); + } else if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_INTERNAL_ERROR || + code == TSDB_CODE_SYN_PROPOSE_NOT_READY || code == TSDB_CODE_RPC_REDIRECT) { + tDebug("code str %s, contlen:%d 1", tstrerror(code), pResp->contLen); + noDelay = cliResetEpset(pCtx, pResp, true); + transFreeMsg(pResp->pCont); + addConnToPool(pThrd->pool, pConn); + } else if (code == TSDB_CODE_SYN_RESTORING) { + tDebug("code str %s, contlen:%d 0", tstrerror(code), pResp->contLen); + noDelay = cliResetEpset(pCtx, pResp, false); + addConnToPool(pThrd->pool, pConn); + transFreeMsg(pResp->pCont); + } else { + tDebug("code str %s, contlen:%d 0", tstrerror(code), pResp->contLen); + noDelay = cliResetEpset(pCtx, pResp, false); + addConnToPool(pThrd->pool, pConn); + transFreeMsg(pResp->pCont); + } + + if (noDelay == false) { + pCtx->epsetRetryCnt = 1; + pCtx->retryStep++; + + int64_t factor = pow(pCtx->retryStepFactor, pCtx->retryStep - 1); + pCtx->retryNextInterval = factor * pCtx->retryMinInterval; + if (pCtx->retryNextInterval >= pCtx->retryMaxInterval) { + pCtx->retryNextInterval = pCtx->retryMaxInterval; + } + + if (-1 != pCtx->retryMaxTimeout && taosGetTimestampMs() - pCtx->retryInitTimestamp >= pCtx->retryMaxTimeout) { + return false; + } + } else { + pCtx->retryNextInterval = 0; + pCtx->epsetRetryCnt++; + } + + pMsg->sent = 0; + cliSchedMsgToNextNode(pMsg, pThrd); + return true; +} int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { SCliThrd* pThrd = pConn->hostThrd; STrans* pTransInst = pThrd->pTransInst; @@ -1431,40 +1555,10 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { STransConnCtx* pCtx = pMsg->ctx; int32_t code = pResp->code; - bool retry = (pTransInst->retry != NULL && pTransInst->retry(code, pResp->msgType - 1)) ? true : false; - if (retry) { - pMsg->sent = 0; - pCtx->retryCnt += 1; - if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK) { - cliCompareAndSwap(&pCtx->retryLimit, pTransInst->retryLimit, EPSET_GET_SIZE(&pCtx->epSet) * 3); - if (pCtx->retryCnt < pCtx->retryLimit) { - transUnrefCliHandle(pConn); - EPSET_FORWARD_INUSE(&pCtx->epSet); - transFreeMsg(pResp->pCont); - cliSchedMsgToNextNode(pMsg, pThrd); - return -1; - } - } else { - cliCompareAndSwap(&pCtx->retryLimit, pTransInst->retryLimit, pTransInst->retryLimit); - if (pCtx->retryCnt < pCtx->retryLimit) { - if (pResp->contLen == 0) { - EPSET_FORWARD_INUSE(&pCtx->epSet); - } else { - if (tDeserializeSEpSet(pResp->pCont, pResp->contLen, &pCtx->epSet) < 0) { - tError("%s conn %p failed to deserialize epset", CONN_GET_INST_LABEL(pConn), pConn); - } - } - addConnToPool(pThrd->pool, pConn); - transFreeMsg(pResp->pCont); - cliSchedMsgToNextNode(pMsg, pThrd); - return -1; - } else { - // change error code for taos client driver if retryCnt exceeds limit - if (0 == strncmp(pTransInst->label, "TSC", strlen("TSC"))) pResp->code = TSDB_CODE_APP_NOT_READY; - } - } + bool retry = cliGenRetryRule(pConn, pResp, pMsg); + if (retry == true) { + return -1; } - STraceId* trace = &pResp->info.traceId; bool hasEpSet = cliTryExtractEpSet(pResp, &pCtx->epSet); diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 395e28d68f..3b41590440 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -1229,7 +1229,7 @@ int transReleaseSrvHandle(void* handle) { m->msg = tmsg; m->type = Release; - tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle); + tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle); if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) { destroySmsg(m); } @@ -1237,11 +1237,11 @@ int transReleaseSrvHandle(void* handle) { transReleaseExHandle(transGetRefMgt(), refId); return 0; _return1: - tTrace("handle %p failed to send to release handle", exh); + tDebug("handle %p failed to send to release handle", exh); transReleaseExHandle(transGetRefMgt(), refId); return -1; _return2: - tTrace("handle %p failed to send to release handle", exh); + tDebug("handle %p failed to send to release handle", exh); return -1; } int transSendResponse(const STransMsg* msg) { @@ -1266,7 +1266,7 @@ int transSendResponse(const STransMsg* msg) { m->type = Normal; STraceId* trace = (STraceId*)&msg->info.traceId; - tGTrace("conn %p start to send resp (1/2)", exh->handle); + tGDebug("conn %p start to send resp (1/2)", exh->handle); if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) { destroySmsg(m); } @@ -1275,12 +1275,12 @@ int transSendResponse(const STransMsg* msg) { return 0; _return1: - tTrace("handle %p failed to send resp", exh); + tDebug("handle %p failed to send resp", exh); rpcFreeCont(msg->pCont); transReleaseExHandle(transGetRefMgt(), refId); return -1; _return2: - tTrace("handle %p failed to send resp", exh); + tDebug("handle %p failed to send resp", exh); rpcFreeCont(msg->pCont); return -1; } @@ -1302,7 +1302,7 @@ int transRegisterMsg(const STransMsg* msg) { m->type = Register; STrans* pTransInst = pThrd->pTransInst; - tTrace("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle); + tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle); if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) { destroySmsg(m); } @@ -1311,12 +1311,12 @@ int transRegisterMsg(const STransMsg* msg) { return 0; _return1: - tTrace("handle %p failed to register brokenlink", exh); + tDebug("handle %p failed to register brokenlink", exh); rpcFreeCont(msg->pCont); transReleaseExHandle(transGetRefMgt(), refId); return -1; _return2: - tTrace("handle %p failed to register brokenlink", exh); + tDebug("handle %p failed to register brokenlink", exh); rpcFreeCont(msg->pCont); return -1; } diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 4d889843e8..1901e48c50 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -95,6 +95,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MSG_DECODE_ERROR, "Msg decode error") TAOS_DEFINE_ERROR(TSDB_CODE_NO_AVAIL_DISK, "No available disk") TAOS_DEFINE_ERROR(TSDB_CODE_NOT_FOUND, "Not found") TAOS_DEFINE_ERROR(TSDB_CODE_NO_DISKSPACE, "Out of disk space") +TAOS_DEFINE_ERROR(TSDB_CODE_TIMEOUT_ERROR, "Operation timeout") //client TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_OPERATION, "Invalid operation") @@ -407,6 +408,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_RECONFIG_NOT_READY, "Sync not ready for re TAOS_DEFINE_ERROR(TSDB_CODE_SYN_PROPOSE_NOT_READY, "Sync not ready for propose") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_STANDBY_NOT_READY, "Sync not ready for standby") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_BATCH_ERROR, "Sync batch error") +TAOS_DEFINE_ERROR(TSDB_CODE_SYN_RESTORING, "Sync is restoring") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR, "Sync internal error") //tq diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 629cc8241f..1df0d5ef71 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -113,7 +113,7 @@ ,,y,script,./test.sh -f tsim/parser/first_last.sim ,,y,script,./test.sh -f tsim/parser/fill_stb.sim ,,y,script,./test.sh -f tsim/parser/interp.sim -,,y,script,./test.sh -f tsim/parser/limit2.sim +#,,y,script,./test.sh -f tsim/parser/limit2.sim ,,y,script,./test.sh -f tsim/parser/fourArithmetic-basic.sim ,,y,script,./test.sh -f tsim/parser/function.sim ,,y,script,./test.sh -f tsim/parser/groupby-basic.sim diff --git a/tests/system-test/0-others/sysinfo.py b/tests/system-test/0-others/sysinfo.py index 4ddae42ac5..9ffa6ca9e6 100644 --- a/tests/system-test/0-others/sysinfo.py +++ b/tests/system-test/0-others/sysinfo.py @@ -48,7 +48,7 @@ class TDTestCase: tdSql.checkData(0,0,1) #!for bug tdDnodes.stoptaosd(1) - sleep(self.delaytime) + sleep(self.delaytime * 5) if platform.system().lower() == 'windows': sleep(10) tdSql.error('select server_status()')