diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt index b4e8825431475c09fbf925671e6d7f691c700b15..de7b75a24520430d74aca7d1a7506acd2e5d7784 100644 --- a/contrib/CMakeLists.txt +++ b/contrib/CMakeLists.txt @@ -27,6 +27,10 @@ else () cat("${TD_SUPPORT_DIR}/taosadapter_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) endif() +if(TD_LINUX_64 AND JEMALLOC_ENABLED) + cat("${TD_SUPPORT_DIR}/jemalloc_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) +endif() + # pthread if(${BUILD_PTHREAD}) cat("${TD_SUPPORT_DIR}/pthread_CMakeLists.txt.in" ${CONTRIB_TMP_FILE}) @@ -392,6 +396,19 @@ if(${BUILD_WITH_SQLITE}) endif(NOT TD_WINDOWS) endif(${BUILD_WITH_SQLITE}) +# jemalloc +IF (TD_LINUX_64 AND JEMALLOC_ENABLED) + include(ExternalProject) + ExternalProject_Add(jemalloc + PREFIX "jemalloc" + SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/jemalloc + BUILD_IN_SOURCE 1 + CONFIGURE_COMMAND ./autogen.sh COMMAND ./configure --prefix=${CMAKE_BINARY_DIR}/build/ + BUILD_COMMAND ${MAKE} + ) + INCLUDE_DIRECTORIES(${CMAKE_BINARY_DIR}/build/include) +ENDIF () + # addr2line if(${BUILD_ADDR2LINE}) if(NOT ${TD_WINDOWS}) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index 7bc2ede4fcdf2430db2e74aa74aaed46a65d124f..a3cc7aaa8ade3dbb2cfb5d04ffd10b15c25159aa 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -123,7 +123,7 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table * @param handle * @return */ -int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds); +int32_t qExecTask(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds); /** * kill the ongoing query and free the query handle and corresponding resources automatically diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index 662558529d29da3d849208c526e65c2db29f6d71..ef9a64e946f2a48908d75f81c7893399c5e08e6a 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -604,28 +604,44 @@ _end: static int32_t tdRSmaFetchAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema, int64_t suid, int8_t blkType) { - while (1) { - SSDataBlock *output = NULL; - uint64_t ts; + SArray *pResList = taosArrayInit(1, POINTER_BYTES); + if (pResList == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } - int32_t code = qExecTask(taskInfo, &output, &ts); + while (1) { + uint64_t ts; + int32_t code = qExecTask(taskInfo, pResList, &ts); if (code < 0) { smaError("vgId:%d, qExecTask for rsma table %" PRIi64 " level %" PRIi8 " failed since %s", SMA_VID(pSma), suid, pItem->level, terrstr(code)); goto _err; } - if (output) { -#if 0 + if (taosArrayGetSize(pResList) == 0) { + if (terrno == 0) { + smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level); + } else { + smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, terrstr()); + goto _err; + } + + break; + } + + for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) { + SSDataBlock *output = taosArrayGetP(pResList, i); + +#if 1 char flag[10] = {0}; snprintf(flag, 10, "level %" PRIi8, pItem->level); - SArray *pResult = taosArrayInit(1, sizeof(SSDataBlock)); - taosArrayPush(pResult, output); - blockDebugShowDataBlocks(pResult, flag); - taosArrayDestroy(pResult); +// blockDebugShowDataBlocks(output, flag); +// taosArrayDestroy(pResult); #endif - STsdb *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]); + STsdb * sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]); SSubmitReq *pReq = NULL; + // TODO: the schema update should be handled later(TD-17965) if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, SMA_VID(pSma), suid) < 0) { smaError("vgId:%d, build submit req for rsma stable %" PRIi64 " level %" PRIi8 " failed since %s", @@ -644,17 +660,14 @@ static int32_t tdRSmaFetchAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSm SMA_VID(pSma), suid, pItem->level, output->info.version); taosMemoryFreeClear(pReq); - } else if (terrno == 0) { - smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level); - break; - } else { - smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, terrstr()); - goto _err; } } + taosArrayDestroy(pResList); return TSDB_CODE_SUCCESS; + _err: + taosArrayDestroy(pResList); return TSDB_CODE_FAILED; } @@ -1407,7 +1420,7 @@ int32_t tdRSmaFetchSend(SSma *pSma, SRSmaInfo *pInfo, int8_t level) { tEncoderClear(&encoder); } tEncoderClear(&encoder); - + ((SMsgHead *)pBuf)->vgId = SMA_VID(pSma); ((SMsgHead *)pBuf)->contLen = contLen + sizeof(SMsgHead); @@ -1434,10 +1447,10 @@ _err: /** * @brief fetch rsma data of level 2/3 and submit - * - * @param pSma - * @param pMsg - * @return int32_t + * + * @param pSma + * @param pMsg + * @return int32_t */ int32_t smaProcessFetch(SSma *pSma, void *pMsg) { SRpcMsg *pRpcMsg = (SRpcMsg *)pMsg; diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index 435bbb77b8cab0b6c631f98e30444501ae8faf03..09a5f1a8c64f087738aa800f56e6b7ed866aff1f 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -80,14 +80,22 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVa } int32_t rowCnt = 0; + SArray* pResList = taosArrayInit(4, POINTER_BYTES); + while (1) { + taosArrayClear(pResList); + SSDataBlock* pDataBlock = NULL; uint64_t ts = 0; tqDebug("task start to execute"); - if (qExecTask(task, &pDataBlock, &ts) < 0) { + if (qExecTask(task, pResList, &ts) < 0) { ASSERT(0); } - tqDebug("task execute end, get %p", pDataBlock); + + if (taosArrayGetSize(pResList) > 0) { + pDataBlock = taosArrayGet(pResList, 0); + tqDebug("task execute end, get %p", pDataBlock); + } if (pDataBlock != NULL) { if (pRsp->withTbName) { @@ -143,6 +151,7 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVa break; } + taosArrayDestroy(pResList); return 0; } diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index 3a37b5e760a637dae05a6f797833b0b9c805bbd2..977adcaa5e3872ff7aeea7209eca7e275ea07398 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -69,10 +69,10 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) { // clang-format off // data format: -// +----------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+ -// |SDataCacheEntry | total length | numOfRows | group id | col1_schema | col2_schema | col3_schema... | column#1 length, column#2 length...| col1 bitmap | col1 data | col2 bitmap | col2 data | .... | | (4 bytes) |(8 bytes) -// | |sizeof(int32) | sizeof(int32)| sizeof(uint64_t) |(sizeof(int16_t)+sizeof(int32_t))*numOfCols | sizeof(int32_t) * numOfCols | actual size | | -// +----------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+ +// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+ +// |SDataCacheEntry | version | total length | numOfRows | group id | col1_schema | col2_schema | col3_schema... | column#1 length, column#2 length...| col1 bitmap | col1 data | col2 bitmap | col2 data | .... | | (4 bytes) |(8 bytes) +// | | sizeof(int32_t) |sizeof(int32) | sizeof(int32)| sizeof(uint64_t) | (sizeof(int8_t)+sizeof(int32_t))*numOfCols | sizeof(int32_t) * numOfCols | actual size | | +// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+ // The length of bitmap is decided by number of rows of this data block, and the length of each column data is // recorded in the first segment, next to the struct header // clang-format on diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 897ddb59ec5e7aa2670d27090b19a62ef3ea83e6..5bbf5928da6c9519a9833c6d844a6b2cf1382f89 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -422,11 +422,17 @@ int waitMoment(SQInfo* pQInfo) { } #endif -int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) { +static void freeBlock(void* param) { + SSDataBlock* pBlock = *(SSDataBlock**) param; + blockDataDestroy(pBlock); +} + +int32_t qExecTask(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; int64_t threadId = taosGetSelfPthreadId(); - *pRes = NULL; + taosArrayClearEx(pResList, freeBlock); + int64_t curOwner = 0; if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) { qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner); @@ -457,23 +463,34 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) { qDebug("%s execTask is launched", GET_TASKID(pTaskInfo)); + int32_t current = 0; + SSDataBlock* pRes = NULL; + int64_t st = taosGetTimestampUs(); - *pRes = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot); + while((pRes = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot)) != NULL) { + SSDataBlock* p = createOneDataBlock(pRes, true); + current += p->info.rows; + ASSERT(p->info.rows > 0); + taosArrayPush(pResList, &p); + + if (current >= 4096) { + break; + } + } + uint64_t el = (taosGetTimestampUs() - st); pTaskInfo->cost.elapsedTime += el; - if (NULL == *pRes) { + if (NULL == pRes) { *useconds = pTaskInfo->cost.elapsedTime; } cleanUpUdfs(); - - int32_t current = (*pRes != NULL) ? (*pRes)->info.rows : 0; uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows; - qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms", - GET_TASKID(pTaskInfo), current, total, 0, el / 1000.0); + qDebug("%s task suspended, %d rows in %d blocks returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms", + GET_TASKID(pTaskInfo), current, (int32_t) taosArrayGetSize(pResList), total, 0, el / 1000.0); atomic_store_64(&pTaskInfo->owner, 0); return pTaskInfo->code; diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index 7c5cfda81fd7d6380777b203c64af44929968652..85a1b0444cd35af1ac9a7d2186dc45dbca48c0e0 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -75,22 +75,20 @@ int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { int32_t code = 0; bool qcontinue = true; - SSDataBlock *pRes = NULL; uint64_t useconds = 0; int32_t i = 0; int32_t execNum = 0; qTaskInfo_t taskHandle = ctx->taskHandle; DataSinkHandle sinkHandle = ctx->sinkHandle; + SArray* pResList = taosArrayInit(4, POINTER_BYTES); while (true) { QW_TASK_DLOG("start to execTask, loopIdx:%d", i++); - pRes = NULL; - // if *taskHandle is NULL, it's killed right now if (taskHandle) { qwDbgSimulateSleep(); - code = qExecTask(taskHandle, &pRes, &useconds); + code = qExecTask(taskHandle, pResList, &useconds); if (code) { if (code != TSDB_CODE_OPS_NOT_SUPPORT) { QW_TASK_ELOG("qExecTask failed, code:%x - %s", code, tstrerror(code)); @@ -103,9 +101,8 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { ++execNum; - if (NULL == pRes) { + if (taosArrayGetSize(pResList) == 0) { QW_TASK_DLOG("qExecTask end with empty res, useconds:%" PRIu64, useconds); - dsEndPut(sinkHandle, useconds); QW_ERR_RET(qwHandleTaskComplete(QW_FPARAMS(), ctx)); @@ -117,19 +114,20 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { break; } - int32_t rows = pRes->info.rows; + for(int32_t j = 0; j < taosArrayGetSize(pResList); ++j) { + SSDataBlock *pRes = taosArrayGetP(pResList, j); + ASSERT(pRes->info.rows > 0); - ASSERT(pRes->info.rows > 0); + SInputData inputData = {.pData = pRes}; + code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue); + if (code) { + QW_TASK_ELOG("dsPutDataBlock failed, code:%x - %s", code, tstrerror(code)); + QW_ERR_RET(code); + } - SInputData inputData = {.pData = pRes}; - code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue); - if (code) { - QW_TASK_ELOG("dsPutDataBlock failed, code:%x - %s", code, tstrerror(code)); - QW_ERR_RET(code); + QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", pRes->info.rows, qcontinue); } - QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", rows, qcontinue); - if (!qcontinue) { if (queryStop) { *queryStop = true; @@ -151,6 +149,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { } } + taosArrayDestroy(pResList); QW_RET(code); } diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 7512f792c1a6a631421f60497a5db95c46d9a788..b92f658de051430d8dafe06b5e3b63c3f89fc3ee 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -43,13 +43,17 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* } // exec + SArray* pResList = taosArrayInit(4, POINTER_BYTES); while (1) { SSDataBlock* output = NULL; uint64_t ts = 0; - if (qExecTask(exec, &output, &ts) < 0) { + + taosArrayClear(pResList); + if (qExecTask(exec, pResList, &ts) < 0) { ASSERT(false); } - if (output == NULL) { + + if (taosArrayGetSize(pResList) == 0) { if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) { SSDataBlock block = {0}; const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)data; @@ -65,6 +69,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* break; } + output = taosArrayGetP(pResList, 0); if (output->info.type == STREAM_RETRIEVE) { if (streamBroadcastToChildren(pTask, output) < 0) { // TODO @@ -79,6 +84,8 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* block.info.childId = pTask->selfChildId; taosArrayPush(pRes, &block); } + + taosArrayDestroy(pResList); return 0; } @@ -98,6 +105,7 @@ int32_t streamPipelineExec(SStreamTask* pTask, int32_t batchNum) { void* exec = pTask->exec.executor; + SArray* pResList = taosArrayInit(4, POINTER_BYTES); while (1) { SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock)); if (pRes == NULL) { @@ -107,14 +115,17 @@ int32_t streamPipelineExec(SStreamTask* pTask, int32_t batchNum) { int32_t batchCnt = 0; while (1) { - SSDataBlock* output = NULL; - uint64_t ts = 0; - if (qExecTask(exec, &output, &ts) < 0) { + uint64_t ts = 0; + taosArrayClear(pResList); + if (qExecTask(exec, pResList, &ts) < 0) { ASSERT(0); } - if (output == NULL) break; + + if (taosArrayGetSize(pResList) == 0) break; SSDataBlock block = {0}; + SSDataBlock* output = taosArrayGetP(pResList, 0); + assignOneDataBlock(&block, output); block.info.childId = pTask->selfChildId; taosArrayPush(pRes, &block);