Merge pull request #29314 from taosdata/fix/TD-33275.2
fix: add more UT cases
This commit is contained in:
commit
f94234e69c
|
@ -424,6 +424,7 @@ int32_t ctgGetTbCfg(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName,
|
|||
CTG_RET(TSDB_CODE_SUCCESS);
|
||||
}
|
||||
|
||||
#if 0
|
||||
int32_t ctgGetTbTag(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName, SArray** pRes) {
|
||||
SVgroupInfo vgroupInfo = {0};
|
||||
STableCfg* pCfg = NULL;
|
||||
|
@ -474,6 +475,7 @@ _return:
|
|||
|
||||
CTG_RET(code);
|
||||
}
|
||||
#endif
|
||||
|
||||
int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName, SArray** pVgList) {
|
||||
STableMeta* tbMeta = NULL;
|
||||
|
@ -1695,6 +1697,7 @@ _return:
|
|||
CTG_API_LEAVE(code);
|
||||
}
|
||||
|
||||
#if 0
|
||||
int32_t catalogGetTableTag(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes) {
|
||||
CTG_API_ENTER();
|
||||
|
||||
|
@ -1709,6 +1712,7 @@ _return:
|
|||
|
||||
CTG_API_LEAVE(code);
|
||||
}
|
||||
#endif
|
||||
|
||||
int32_t catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, STableCfg** pCfg) {
|
||||
CTG_API_ENTER();
|
||||
|
@ -1845,7 +1849,7 @@ _return:
|
|||
CTG_API_LEAVE(code);
|
||||
}
|
||||
|
||||
|
||||
#if 0
|
||||
int32_t catalogAsyncUpdateViewMeta(SCatalog* pCtg, SViewMetaRsp* pMsg) {
|
||||
CTG_API_ENTER();
|
||||
|
||||
|
@ -1860,6 +1864,7 @@ _return:
|
|||
|
||||
CTG_API_LEAVE(code);
|
||||
}
|
||||
#endif
|
||||
|
||||
int32_t catalogGetViewMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pViewName, STableMeta** pTableMeta) {
|
||||
CTG_API_ENTER();
|
||||
|
|
|
@ -2992,6 +2992,7 @@ TEST(apiTest, catalogGetTableIndex_test) {
|
|||
catalogDestroy();
|
||||
}
|
||||
|
||||
|
||||
TEST(apiTest, catalogGetDBCfg_test) {
|
||||
struct SCatalog *pCtg = NULL;
|
||||
SRequestConnInfo connInfo = {0};
|
||||
|
|
|
@ -41,6 +41,7 @@
|
|||
#include "tvariant.h"
|
||||
#include "stub.h"
|
||||
#include "querytask.h"
|
||||
#include "hashjoin.h"
|
||||
|
||||
|
||||
namespace {
|
||||
|
@ -3766,6 +3767,21 @@ TEST(leftWinJoin, noCondProjectionTest) {
|
|||
|
||||
#endif
|
||||
|
||||
#if 1
|
||||
TEST(functionsTest, branch) {
|
||||
struct SOperatorInfo op = {0};
|
||||
SHJoinOperatorInfo join;
|
||||
SBufRowInfo bufrow = {0};
|
||||
SSDataBlock blk = {0};
|
||||
|
||||
op.info = &join;
|
||||
memset(&join, 0, sizeof(join));
|
||||
join.ctx.pBuildRow = &bufrow;
|
||||
blk.info.rows = 1;
|
||||
join.finBlk = &blk;
|
||||
hInnerJoinDo(&op);
|
||||
}
|
||||
#endif
|
||||
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
|
|
@ -161,6 +161,10 @@ int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
|
|||
int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan) { return nodesStringToNode(pStr, (SNode**)pSubplan); }
|
||||
|
||||
int32_t qSubPlanToMsg(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
|
||||
if (NULL == pSubplan) {
|
||||
return terrno = TSDB_CODE_INVALID_PARA;
|
||||
}
|
||||
|
||||
if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {
|
||||
SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;
|
||||
*pLen = insert->size;
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
#include "qwInt.h"
|
||||
#include "qworker.h"
|
||||
|
||||
#if 0
|
||||
void qwSetConcurrentTaskNumCb(int32_t taskNum) {
|
||||
int32_t finTaskNum = TMIN(taskNum, tsNumOfQueryThreads * QW_DEFAULT_THREAD_TASK_NUM);
|
||||
|
||||
|
@ -33,6 +34,7 @@ void qwIncConcurrentTaskNumCb(void) {
|
|||
|
||||
//TODO
|
||||
}
|
||||
#endif
|
||||
|
||||
int32_t qwInitJobInfo(QW_FPARAMS_DEF, SQWJobInfo* pJob) {
|
||||
pJob->pSessions= taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
|
||||
|
|
|
@ -149,11 +149,11 @@ typedef struct SSchedulerMgmt {
|
|||
bool exit;
|
||||
int32_t jobRef;
|
||||
int32_t jobNum;
|
||||
SSchStat stat;
|
||||
void *timer;
|
||||
SRWLatch hbLock;
|
||||
SHashObj *hbConnections;
|
||||
void *queryMgmt;
|
||||
SSchStat stat;
|
||||
} SSchedulerMgmt;
|
||||
|
||||
typedef struct SSchCallbackParamHeader {
|
||||
|
|
|
@ -50,11 +50,6 @@ int32_t schChkJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) {
|
|||
int32_t taskNum = taosArrayGetSize(pJob->dataSrcTasks);
|
||||
for (int32_t i = 0; i < taskNum; ++i) {
|
||||
SSchTask *pTask = *(SSchTask **)taosArrayGet(pJob->dataSrcTasks, i);
|
||||
if (NULL == pTask) {
|
||||
SCH_JOB_DLOG("fail to get the %dth task", i);
|
||||
SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
sum += pTask->plan->execNodeStat.tableNum;
|
||||
}
|
||||
|
||||
|
|
|
@ -273,18 +273,18 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
|
|||
if (taskDone < pTask->level->taskNum) {
|
||||
SCH_TASK_DLOG("wait all tasks, done:%d, all:%d", taskDone, pTask->level->taskNum);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
} else if (taskDone > pTask->level->taskNum) {
|
||||
SCH_TASK_ELOG("taskDone number invalid, done:%d, total:%d", taskDone, pTask->level->taskNum);
|
||||
}
|
||||
|
||||
SCH_TASK_DLOG("taskDone number reach level task number, done:%d, total:%d", taskDone, pTask->level->taskNum);
|
||||
|
||||
if (pTask->level->taskFailed > 0) {
|
||||
SCH_RET(schHandleJobFailure(pJob, pJob->errCode));
|
||||
} else {
|
||||
SCH_RET(schSwitchJobStatus(pJob, JOB_TASK_STATUS_PART_SUCC, NULL));
|
||||
}
|
||||
} else {
|
||||
pJob->resNode = pTask->succeedAddr;
|
||||
|
||||
SCH_RET(schSwitchJobStatus(pJob, JOB_TASK_STATUS_PART_SUCC, NULL));
|
||||
}
|
||||
|
||||
pJob->resNode = pTask->succeedAddr;
|
||||
|
||||
pJob->fetchTask = pTask;
|
||||
|
||||
|
@ -600,12 +600,12 @@ int32_t schPushTaskToExecList(SSchJob *pJob, SSchTask *pTask) {
|
|||
int32_t code = taosHashPut(pJob->execTasks, &pTask->taskId, sizeof(pTask->taskId), &pTask, POINTER_BYTES);
|
||||
if (0 != code) {
|
||||
if (HASH_NODE_EXIST(code)) {
|
||||
SCH_TASK_DLOG("task already in execTask list, code:%x", code);
|
||||
SCH_TASK_DLOG("task already in execTask list, code:0x%x", code);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
SCH_TASK_ELOG("taosHashPut task to execTask list failed, errno:0x%x", errno);
|
||||
SCH_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
SCH_TASK_ELOG("taosHashPut task to execTask list failed, code:0x%x", code);
|
||||
SCH_ERR_RET(code);
|
||||
}
|
||||
|
||||
SCH_TASK_DLOG("task added to execTask list, numOfTasks:%d", taosHashGetSize(pJob->execTasks));
|
||||
|
@ -800,11 +800,6 @@ int32_t schSetAddrsFromNodeList(SSchJob *pJob, SSchTask *pTask) {
|
|||
|
||||
for (int32_t i = 0; i < nodeNum; ++i) {
|
||||
SQueryNodeLoad *nload = taosArrayGet(pJob->nodeList, i);
|
||||
if (NULL == nload) {
|
||||
SCH_TASK_ELOG("fail to get the %dth node in nodeList, nodeNum:%d", i, nodeNum);
|
||||
SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
SQueryNodeAddr *naddr = &nload->addr;
|
||||
|
||||
if (NULL == taosArrayPush(pTask->candidateAddrs, naddr)) {
|
||||
|
|
|
@ -60,6 +60,19 @@ extern "C" int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t
|
|||
extern "C" int32_t schHandleNotifyCallback(void *param, SDataBuf *pMsg, int32_t code);
|
||||
extern "C" int32_t schHandleLinkBrokenCallback(void *param, SDataBuf *pMsg, int32_t code);
|
||||
extern "C" int32_t schRescheduleTask(SSchJob *pJob, SSchTask *pTask);
|
||||
extern "C" int32_t schValidateRspMsgType(SSchJob *pJob, SSchTask *pTask, int32_t msgType);
|
||||
//extern "C" int32_t schProcessFetchRsp(SSchJob *pJob, SSchTask *pTask, char *msg, int32_t rspCode);
|
||||
extern "C" int32_t schProcessResponseMsg(SSchJob *pJob, SSchTask *pTask, SDataBuf *pMsg, int32_t rspCode);
|
||||
extern "C" void schInitTaskRetryTimes(SSchJob *pJob, SSchTask *pTask, SSchLevel *pLevel);
|
||||
extern "C" int32_t schRecordTaskSucceedNode(SSchJob *pJob, SSchTask *pTask);
|
||||
extern "C" int32_t schDropTaskExecNode(SSchJob *pJob, SSchTask *pTask, void *handle, int32_t execId);
|
||||
extern "C" int32_t schPushTaskToExecList(SSchJob *pJob, SSchTask *pTask);
|
||||
extern "C" int32_t schSetAddrsFromNodeList(SSchJob *pJob, SSchTask *pTask);
|
||||
extern "C" int32_t schNotifyTaskOnExecNode(SSchJob *pJob, SSchTask *pTask, ETaskNotifyType type);
|
||||
extern "C" int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask);
|
||||
extern "C" int32_t schLaunchTaskImpl(void *param);
|
||||
extern "C" void schHandleTimerEvent(void *param, void *tmrId);
|
||||
|
||||
|
||||
int64_t insertJobRefId = 0;
|
||||
int64_t queryJobRefId = 0;
|
||||
|
@ -1473,6 +1486,263 @@ TEST(otherTest, function) {
|
|||
schMgmt.jobRef = -1;
|
||||
}
|
||||
|
||||
TEST(otherTest, branch) {
|
||||
SSchJob job = {0};
|
||||
SSchTask task = {0};
|
||||
memset(&schMgmt, 0, sizeof(schMgmt));
|
||||
|
||||
schValidateRspMsgType(&job, &task, TDMT_SCH_MERGE_FETCH_RSP);
|
||||
|
||||
task.lastMsgType = TDMT_SCH_MERGE_FETCH_RSP - 1;
|
||||
schValidateRspMsgType(&job, &task, TDMT_SCH_MERGE_FETCH_RSP);
|
||||
|
||||
schValidateRspMsgType(&job, &task, 0);
|
||||
|
||||
schValidateRspMsgType(&job, &task, TDMT_SCH_QUERY_RSP);
|
||||
|
||||
task.lastMsgType = TDMT_SCH_QUERY_RSP - 1;
|
||||
schValidateRspMsgType(&job, &task, TDMT_SCH_QUERY_RSP);
|
||||
|
||||
schProcessFetchRsp(&job, &task, NULL, -1);
|
||||
schProcessFetchRsp(&job, &task, NULL, 0);
|
||||
|
||||
job.fetchRes = (void*)0x1;
|
||||
schProcessFetchRsp(&job, &task, (char*)taosMemoryMalloc(0), 0);
|
||||
job.fetchRes = NULL;
|
||||
|
||||
SDataBuf databuf = {0};
|
||||
databuf.msgType = TDMT_VND_ALTER_TABLE_RSP;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_VND_SUBMIT_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_VND_DELETE_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_SCH_QUERY_RSP;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_SCH_QUERY_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
|
||||
databuf.msgType = TDMT_SCH_EXPLAIN_RSP;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_SCH_EXPLAIN_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
job.status = 0;
|
||||
|
||||
job.attr.explainMode = EXPLAIN_MODE_ANALYZE;
|
||||
databuf.msgType = TDMT_SCH_EXPLAIN_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
job.fetchRes = (void*)0x1;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
job.fetchRes = NULL;
|
||||
|
||||
job.attr.explainMode = EXPLAIN_MODE_ANALYZE;
|
||||
databuf.msgType = TDMT_SCH_EXPLAIN_RSP;
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_SCH_DROP_TASK_RSP;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = TDMT_SCH_LINK_BROKEN;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.msgType = 0;
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessResponseMsg(&job, &task, &databuf, 0);
|
||||
|
||||
databuf.pData = taosMemoryMalloc(0);
|
||||
schHandleHbCallback(NULL, &databuf, 0);
|
||||
|
||||
__async_send_cb_fn_t fp = NULL;
|
||||
schGetCallbackFp(TDMT_SCH_TASK_NOTIFY, &fp);
|
||||
schGetCallbackFp(0, &fp);
|
||||
|
||||
SQueryNodeEpId ep = {0};
|
||||
schBuildAndSendHbMsg(&ep, NULL);
|
||||
|
||||
schBuildAndSendMsg(&job, &task, NULL, 0, NULL);
|
||||
|
||||
SSchLevel level = {0};
|
||||
SSubplan subplan;
|
||||
memset(&subplan, 0, sizeof(subplan));
|
||||
job.attr.queryJob = true;
|
||||
schMgmt.cfg.schPolicy = SCH_ALL;
|
||||
task.plan = &subplan;
|
||||
schInitTaskRetryTimes(&job, &task, &level);
|
||||
job.attr.queryJob = false;
|
||||
memset(&schMgmt.cfg, 0, sizeof(schMgmt.cfg));
|
||||
memset(&level, 0, sizeof(level));
|
||||
|
||||
schRecordTaskSucceedNode(&job, &task);
|
||||
|
||||
schDropTaskExecNode(&job, &task, NULL, 0);
|
||||
|
||||
task.execNodes = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
|
||||
schDropTaskExecNode(&job, &task, NULL, 0);
|
||||
|
||||
int32_t execId = 0;
|
||||
task.execId = 1;
|
||||
(void)taosHashPut(task.execNodes, &execId, sizeof(execId), &execId, sizeof(execId));
|
||||
schDropTaskExecNode(&job, &task, NULL, execId);
|
||||
task.execId = 0;
|
||||
taosHashCleanup(task.execNodes);
|
||||
task.execNodes = NULL;
|
||||
|
||||
job.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessOnTaskFailure(&job, &task, 0);
|
||||
job.status = 0;
|
||||
|
||||
task.status = JOB_TASK_STATUS_FAIL;
|
||||
schProcessOnTaskFailure(&job, &task, 0);
|
||||
task.status = 0;
|
||||
|
||||
task.level = &level;
|
||||
schProcessOnTaskFailure(&job, &task, TSDB_CODE_SCH_TIMEOUT_ERROR);
|
||||
memset(&level, 0, sizeof(level));
|
||||
task.level = NULL;
|
||||
|
||||
subplan.subplanType = SUBPLAN_TYPE_SCAN;
|
||||
task.plan = &subplan;
|
||||
SEpSet epset = {0};
|
||||
epset.numOfEps = 127;
|
||||
schChkUpdateRedirectCtx(&job, &task, &epset, 0);
|
||||
|
||||
schChkUpdateRedirectCtx(&job, &task, NULL, 0);
|
||||
task.plan = NULL;
|
||||
|
||||
schPushTaskToExecList(&job, &task);
|
||||
|
||||
job.execTasks = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
|
||||
taosHashPut(job.execTasks, &task.taskId, sizeof(task.taskId), &task, POINTER_BYTES);
|
||||
schPushTaskToExecList(&job, &task);
|
||||
taosHashCleanup(job.execTasks);
|
||||
job.execTasks = NULL;
|
||||
|
||||
bool needRetry = false;
|
||||
task.timeoutUsec = SCH_MAX_TASK_TIMEOUT_USEC / 2 + 1;
|
||||
task.retryTimes = 0;
|
||||
task.maxRetryTimes = 0;
|
||||
schTaskCheckSetRetry(&job, &task, TSDB_CODE_SCH_TIMEOUT_ERROR, &needRetry);
|
||||
|
||||
task.execId = 0;
|
||||
task.retryTimes = 0;
|
||||
task.maxRetryTimes = 100;
|
||||
task.maxExecTimes = 1;
|
||||
schTaskCheckSetRetry(&job, &task, TSDB_CODE_SCH_TIMEOUT_ERROR, &needRetry);
|
||||
|
||||
|
||||
task.execId = 0;
|
||||
task.retryTimes = 0;
|
||||
task.maxRetryTimes = 100;
|
||||
task.maxExecTimes = 100;
|
||||
task.lastMsgType = TDMT_SCH_LINK_BROKEN;
|
||||
schTaskCheckSetRetry(&job, &task, TSDB_CODE_SCH_TIMEOUT_ERROR, &needRetry);
|
||||
|
||||
schSetAddrsFromNodeList(&job, &task);
|
||||
|
||||
schSwitchTaskCandidateAddr(&job, &task);
|
||||
|
||||
|
||||
task.candidateAddrs = taosArrayInit(SCHEDULE_DEFAULT_MAX_NODE_NUM, sizeof(SQueryNodeAddr));
|
||||
SQueryNodeAddr addr = {0};
|
||||
taosArrayPush(task.candidateAddrs, &addr);
|
||||
taosArrayPush(task.candidateAddrs, &addr);
|
||||
schMgmt.cfg.schPolicy = SCH_LOAD_SEQ;
|
||||
task.candidateIdx = 1;
|
||||
schSwitchTaskCandidateAddr(&job, &task);
|
||||
|
||||
schMgmt.cfg.schPolicy = SCH_RANDOM;
|
||||
schSwitchTaskCandidateAddr(&job, &task);
|
||||
taosArrayDestroy(task.candidateAddrs);
|
||||
task.candidateAddrs = NULL;
|
||||
memset(&schMgmt.cfg, 0, sizeof(schMgmt.cfg));
|
||||
task.candidateIdx = 0;
|
||||
|
||||
schDropTaskOnExecNode(&job, &task);
|
||||
|
||||
schNotifyTaskOnExecNode(&job, &task, TASK_NOTIFY_FINISHED);
|
||||
|
||||
schLaunchRemoteTask(&job, &task);
|
||||
|
||||
SSchTaskCtx* pCtx = (SSchTaskCtx*)taosMemoryCalloc(1, sizeof(SSchTaskCtx));
|
||||
pCtx->jobRid = -1;
|
||||
schLaunchTaskImpl((void*)pCtx);
|
||||
|
||||
task.plan = &subplan;
|
||||
subplan.subplanType = SUBPLAN_TYPE_SCAN;
|
||||
job.attr.needFlowCtrl = true;
|
||||
level.taskNum = 1000;
|
||||
task.level = &level;
|
||||
schLaunchTask(&job, &task);
|
||||
task.plan = NULL;
|
||||
task.level = NULL;
|
||||
job.attr.needFlowCtrl = false;
|
||||
|
||||
SSchTimerParam param = {0};
|
||||
param.rId = -1;
|
||||
schHandleTimerEvent(¶m, NULL);
|
||||
|
||||
job.execTasks = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
|
||||
task.delayExecMs = 1;
|
||||
schMgmt.timer = NULL;
|
||||
schDelayLaunchTask(&job, &task);
|
||||
task.delayExecMs = 0;
|
||||
taosHashCleanup(job.execTasks);
|
||||
job.execTasks = NULL;
|
||||
|
||||
job.fetchRes = (void*)0x1;
|
||||
schLaunchFetchTask(&job);
|
||||
job.fetchRes = NULL;
|
||||
|
||||
job.fetchTask = &task;
|
||||
job.attr.localExec = true;
|
||||
job.attr.queryJob = true;
|
||||
subplan.subplanType = SUBPLAN_TYPE_MERGE;
|
||||
task.plan = &subplan;
|
||||
void* p = taosMemoryCalloc(1, 1024);
|
||||
schMgmt.queryMgmt = p;
|
||||
schLaunchFetchTask(&job);
|
||||
memset(&job, 0, sizeof(job));
|
||||
memset(&subplan, 0, sizeof(subplan));
|
||||
task.plan = NULL;
|
||||
taosMemoryFreeClear(schMgmt.queryMgmt);
|
||||
|
||||
// flow ctrl
|
||||
|
||||
job.flowCtrl = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
|
||||
SEp sep = {0};
|
||||
SSchFlowControl nctrl = {0};
|
||||
nctrl.taskList = taosArrayInit(1, POINTER_BYTES);
|
||||
taosHashPut(job.flowCtrl, &sep, sizeof(SEp), &nctrl, sizeof(nctrl));
|
||||
schFreeFlowCtrl(&job);
|
||||
|
||||
|
||||
|
||||
schMgmt.jobRef = -1;
|
||||
}
|
||||
|
||||
|
||||
void schtReset() {
|
||||
insertJobRefId = 0;
|
||||
queryJobRefId = 0;
|
||||
|
|
|
@ -2048,7 +2048,252 @@ TEST(DisablePoolFuncTest, MultiThreadTest) {
|
|||
}
|
||||
#endif
|
||||
|
||||
#if 1
|
||||
TEST(functionsTest, internalFunc) {
|
||||
char* caseName = "functionsTest:internalFunc";
|
||||
int32_t code = 0;
|
||||
|
||||
int64_t msize = 10;
|
||||
void* pSession = NULL;
|
||||
void* pJob = NULL;
|
||||
|
||||
mptInitPool();
|
||||
|
||||
memset(mptCtx.jobCtxs, 0, sizeof(*mptCtx.jobCtxs));
|
||||
|
||||
assert(0 == taosMemPoolCallocJob(0, 0, (void**)&pJob));
|
||||
assert(0 == taosMemPoolInitSession(gMemPoolHandle, &pSession, pJob, "id"));
|
||||
|
||||
int32_t loopTimes = 1;
|
||||
int64_t st = 0;
|
||||
void **addrList = (void**)taosMemCalloc(loopTimes, POINTER_BYTES);
|
||||
|
||||
|
||||
// MALLOC
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMalloc(msize);
|
||||
}
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMalloc(msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMalloc(msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
// CALLOC
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryCalloc(1, msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryCalloc(1, msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryCalloc(1, msize);
|
||||
}
|
||||
//mptFreeAddrList(addrList, loopTimes); NO FREE FOR REALLOC
|
||||
|
||||
// REALLOC
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryRealloc(addrList[i], msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryRealloc(addrList[i], msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryRealloc(addrList[i], msize);
|
||||
}
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
// STRDUP
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrdup("abc");
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrdup("abc");
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrdup("abc");
|
||||
}
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
// STRNDUP
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrndup("abc", 3);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrndup("abc", 3);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptStrndup("abc", 3);
|
||||
}
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
// ALIGNALLOC
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMallocAlign(8, msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMallocAlign(8, msize);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
mptFreeAddrList(addrList, loopTimes);
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMallocAlign(8, msize);
|
||||
}
|
||||
//mptFreeAddrList(addrList, loopTimes); NO FREE FOR GETSIZE
|
||||
|
||||
|
||||
// GETSIZE
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemorySize(addrList[i]);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemorySize(addrList[i]);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemorySize(addrList[i]);
|
||||
}
|
||||
|
||||
// FREE
|
||||
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemoryFree(addrList[i]);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMalloc(msize);
|
||||
}
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemoryFree(addrList[i]);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
addrList[i] = (char*)mptMemoryMalloc(msize);
|
||||
}
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemoryFree(addrList[i]);
|
||||
}
|
||||
|
||||
// TRIM
|
||||
|
||||
bool trimed = false;
|
||||
tsMemPoolFullFunc = 0;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemoryTrim(0, NULL);
|
||||
mptMemoryTrim(0, &trimed);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
tsMemPoolFullFunc = 1;
|
||||
mptEnableMemoryPoolUsage(gMemPoolHandle, pSession);
|
||||
for (int32_t i = 0; i < loopTimes; ++i) {
|
||||
mptMemoryTrim(0, NULL);
|
||||
mptMemoryTrim(0, &trimed);
|
||||
}
|
||||
mptDisableMemoryPoolUsage();
|
||||
|
||||
|
||||
}
|
||||
#endif
|
||||
|
||||
|
||||
#endif
|
||||
|
|
Loading…
Reference in New Issue