diff --git a/include/common/tcommon.h b/include/common/tcommon.h index d88747eb928986ab0e242901bb354b7fd9e77493..72d969a0d54be5c547d984e5b80750093180fcd7 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -195,7 +195,7 @@ typedef struct SDataBlockInfo { uint32_t capacity; SBlockID id; int16_t hasVarCol; - int16_t dataLoad; // denote if the data is loaded or not + int16_t dataLoad; // denote if the data is loaded or not // TODO: optimize and remove following int64_t version; // used for stream, and need serialization @@ -204,8 +204,8 @@ typedef struct SDataBlockInfo { STimeWindow calWin; // used for stream, do not serialize TSKEY watermark; // used for stream - char parTbName[TSDB_TABLE_NAME_LEN]; // used for stream partition - STag* pTag; // used for stream partition + char parTbName[TSDB_TABLE_NAME_LEN]; // used for stream partition + STag* pTag; // used for stream partition } SDataBlockInfo; typedef struct SSDataBlock { @@ -239,22 +239,22 @@ typedef struct SVarColAttr { // pBlockAgg->numOfNull == info.rows, all data are null // pBlockAgg->numOfNull == 0, no data are null. typedef struct SColumnInfoData { - char* pData; // the corresponding block data in memory + char* pData; // the corresponding block data in memory union { char* nullbitmap; // bitmap, one bit for each item in the list SVarColAttr varmeta; }; - SColumnInfo info; // column info - bool hasNull; // if current column data has null value. + SColumnInfo info; // column info + bool hasNull; // if current column data has null value. } SColumnInfoData; typedef struct SQueryTableDataCond { uint64_t suid; - int32_t order; // desc|asc order to iterate the data block + int32_t order; // desc|asc order to iterate the data block int32_t numOfCols; SColumnInfo* colList; - int32_t* pSlotList; // the column output destation slot, and it may be null - int32_t type; // data block load type: + int32_t* pSlotList; // the column output destation slot, and it may be null + int32_t type; // data block load type: STimeWindow twindows; int64_t startVersion; int64_t endVersion; @@ -300,6 +300,7 @@ typedef struct STableBlockDistInfo { int32_t firstSeekTimeUs; uint32_t numOfInmemRows; uint32_t numOfSmallBlocks; + uint32_t numOfVgroups; int32_t blockRowsHisto[20]; } STableBlockDistInfo; diff --git a/include/util/tworker.h b/include/util/tworker.h index 0636f16dbbd50b709de18ce556d8a9dcbdcfba8d..8508adf052bf8eb368dd9846d0a5e1ada1bb6e6e 100644 --- a/include/util/tworker.h +++ b/include/util/tworker.h @@ -26,12 +26,12 @@ extern "C" { typedef struct SQWorkerPool SQWorkerPool; typedef struct SWWorkerPool SWWorkerPool; -typedef struct SQWorker { +typedef struct SQueueWorker { int32_t id; // worker id int64_t pid; // thread pid TdThread thread; // thread id void *pool; -} SQWorker; +} SQueueWorker; typedef struct SQWorkerPool { int32_t max; // max number of workers @@ -39,7 +39,7 @@ typedef struct SQWorkerPool { int32_t num; // current number of workers STaosQset *qset; const char *name; - SQWorker *workers; + SQueueWorker *workers; TdThreadMutex mutex; } SQWorkerPool; diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 47ed2cf035a13314b81f2208718d2c466efb5c68..b01a8717027a06481168759200a89a4adea7fc0a 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -347,7 +347,7 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) { continue; } - if (pRequest->killed) { + if (pRequest->killed || 0 == pRequest->body.queryJob) { releaseRequest(*rid); pIter = taosHashIterate(pObj->pRequests, pIter); continue; diff --git a/source/dnode/mnode/impl/inc/mndInt.h b/source/dnode/mnode/impl/inc/mndInt.h index 785ecc2bf502ca5f2a1e1203f1dc5fbe29b60a21..d851578a744f740cf0e211ee6928b2ec3f0dd705 100644 --- a/source/dnode/mnode/impl/inc/mndInt.h +++ b/source/dnode/mnode/impl/inc/mndInt.h @@ -58,7 +58,7 @@ typedef int32_t (*MndInitFp)(SMnode *pMnode); typedef void (*MndCleanupFp)(SMnode *pMnode); typedef int32_t (*ShowRetrieveFp)(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows); typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter); -typedef struct SQWorker SQHandle; +typedef struct SQueueWorker SQHandle; typedef struct { const char *name; diff --git a/source/dnode/qnode/inc/qndInt.h b/source/dnode/qnode/inc/qndInt.h index e8ccb75040cbc54ab080e7887a877ce9ff67006b..86deda52ad242e371ab59542532234778e26864d 100644 --- a/source/dnode/qnode/inc/qndInt.h +++ b/source/dnode/qnode/inc/qndInt.h @@ -29,7 +29,7 @@ extern "C" { #endif -typedef struct SQWorker SQHandle; +typedef struct SQueueWorker SQHandle; typedef struct SQnode { int32_t qndId; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index ec925087d09c86fe60204c18457e2ea7b6d6823d..adec0272840870a0498adccd7c67036754752772 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -58,7 +58,7 @@ typedef struct STQ STQ; typedef struct SVState SVState; typedef struct SVStatis SVStatis; typedef struct SVBufPool SVBufPool; -typedef struct SQWorker SQHandle; +typedef struct SQueueWorker SQHandle; typedef struct STsdbKeepCfg STsdbKeepCfg; typedef struct SMetaSnapReader SMetaSnapReader; typedef struct SMetaSnapWriter SMetaSnapWriter; diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index 88d9fb1a1f5b4a4f52fead820c4b00981d64e01e..6516c2c920f702b352c4161ff423571e3a7059a8 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -81,10 +81,10 @@ typedef struct MergeIndex { } MergeIndex; typedef struct SBlockDistInfo { - SSDataBlock* pResBlock; - STsdbReader* pHandle; - SReadHandle readHandle; - uint64_t uid; // table uid + SSDataBlock* pResBlock; + STsdbReader* pHandle; + SReadHandle readHandle; + uint64_t uid; // table uid } SBlockDistInfo; static int32_t sysChkFilter__Comm(SNode* pNode); @@ -129,10 +129,10 @@ static char* SYSTABLE_IDX_COLUMN[] = {"table_name", "db_name", "create_time" static char* SYSTABLE_SPECIAL_COL[] = {"db_name", "vgroup_id"}; -static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity); -static SSDataBlock* buildInfoSchemaTableMetaBlock(char* tableName); -static void destroySysScanOperator(void* param); -static int32_t loadSysTableCallback(void* param, SDataBuf* pMsg, int32_t code); +static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity); +static SSDataBlock* buildInfoSchemaTableMetaBlock(char* tableName); +static void destroySysScanOperator(void* param); +static int32_t loadSysTableCallback(void* param, SDataBuf* pMsg, int32_t code); static __optSysFilter optSysGetFilterFunc(int32_t ctype, bool* reverse); static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo, SMetaReader* smrSuperTable, @@ -199,11 +199,11 @@ int32_t sysFilte__TableName(void* arg, SNode* pNode, SArray* result) { if (func == NULL) return -1; SMetaFltParam param = {.suid = 0, - .cid = 0, - .type = TSDB_DATA_TYPE_VARCHAR, - .val = pVal->datum.p, - .reverse = reverse, - .filterFunc = func}; + .cid = 0, + .type = TSDB_DATA_TYPE_VARCHAR, + .val = pVal->datum.p, + .reverse = reverse, + .filterFunc = func}; return -1; } @@ -218,11 +218,11 @@ int32_t sysFilte__CreateTime(void* arg, SNode* pNode, SArray* result) { if (func == NULL) return -1; SMetaFltParam param = {.suid = 0, - .cid = 0, - .type = TSDB_DATA_TYPE_BIGINT, - .val = &pVal->datum.i, - .reverse = reverse, - .filterFunc = func}; + .cid = 0, + .type = TSDB_DATA_TYPE_BIGINT, + .val = &pVal->datum.i, + .reverse = reverse, + .filterFunc = func}; int32_t ret = metaFilterCreateTime(pMeta, ¶m, result); return ret; @@ -350,9 +350,9 @@ static int32_t optSysMergeRslt(SArray* mRslt, SArray* rslt); static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableScanInfo* pInfo, const char* name, SExecTaskInfo* pTaskInfo); void extractTbnameSlotId(SSysTableScanInfo* pInfo, const SScanPhysiNode* pScanNode); -static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, - const char* name, SSDataBlock* pBlock); -__optSysFilter optSysGetFilterFunc(int32_t ctype, bool* reverse) { +static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, const char* name, + SSDataBlock* pBlock); +__optSysFilter optSysGetFilterFunc(int32_t ctype, bool* reverse) { if (ctype == OP_TYPE_LOWER_EQUAL || ctype == OP_TYPE_LOWER_THAN) { *reverse = true; } @@ -516,9 +516,10 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) { metaTbCursorPrev(pInfo->pCur); blockFull = true; } else { - sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows, dataBlock); + sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows, + dataBlock); } - + metaReaderClear(&smrSuperTable); if (blockFull || numOfRows >= pOperator->resultInfo.capacity) { @@ -1343,7 +1344,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { setOperatorCompleted(pOperator); } - return pBlock->info.rows > 0? pBlock:NULL; + return pBlock->info.rows > 0 ? pBlock : NULL; } else { return NULL; } @@ -1357,7 +1358,7 @@ static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScan if (pInfo->tbnameSlotId != -1) { SColumnInfoData* pColumnInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, pInfo->tbnameSlotId); - char varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0}; + char varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0}; memcpy(varDataVal(varTbName), name, strlen(name)); varDataSetLen(varTbName, strlen(name)); @@ -1489,10 +1490,11 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan setOperatorInfo(pOperator, "SysTableScanOperator", QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo); pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock); - pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, optrDefaultBufFn, NULL); + pOperator->fpSet = + createOperatorFpSet(optrDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, optrDefaultBufFn, NULL); return pOperator; - _error: +_error: if (pInfo != NULL) { destroySysScanOperator(pInfo); } @@ -1927,7 +1929,7 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) { // make the valgrind happy that all memory buffer has been initialized already. if (slotId != 0) { SColumnInfoData* p1 = taosArrayGet(pBlock->pDataBlock, 0); - int64_t v = 0; + int64_t v = 0; colDataAppendInt64(p1, 0, &v); } @@ -1937,10 +1939,10 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) { } static void destroyBlockDistScanOperatorInfo(void* param) { - SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param; - blockDataDestroy(pDistInfo->pResBlock); - tsdbReaderClose(pDistInfo->pHandle); - taosMemoryFreeClear(param); + SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param; + blockDataDestroy(pDistInfo->pResBlock); + tsdbReaderClose(pDistInfo->pHandle); + taosMemoryFreeClear(param); } static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pCond) { @@ -2001,7 +2003,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi } pInfo->readHandle = *readHandle; - pInfo->uid = (pBlockScanNode->suid != 0)? pBlockScanNode->suid:pBlockScanNode->uid; + pInfo->uid = (pBlockScanNode->suid != 0) ? pBlockScanNode->suid : pBlockScanNode->uid; int32_t numOfCols = 0; SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols); @@ -2012,8 +2014,8 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo); - pOperator->fpSet = - createOperatorFpSet(optrDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, optrDefaultBufFn, NULL); + pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, + optrDefaultBufFn, NULL); return pOperator; _error: diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 50ceda4605393b224eac68553091c3bc320f166e..e42812db56f0faf96673b2d6d066d4c20f8c9416 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -775,13 +775,13 @@ int32_t maxFunction(SqlFunctionCtx* pCtx) { static int32_t setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t rowIndex); static int32_t setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos, - int32_t rowIndex); + int32_t rowIndex); int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t code = TSDB_CODE_SUCCESS; SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx); - SMinmaxResInfo* pRes = GET_ROWCELL_INTERBUF(pEntryInfo); + SMinmaxResInfo* pRes = GET_ROWCELL_INTERBUF(pEntryInfo); int32_t slotId = pCtx->pExpr->base.resSchema.slotId; int32_t currentRow = pBlock->info.rows; @@ -795,7 +795,7 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_BIGINT: ((int64_t*)pCol->pData)[currentRow] = pRes->v; -// colDataAppendInt64(pCol, currentRow, &pRes->v); + // colDataAppendInt64(pCol, currentRow, &pRes->v); break; case TSDB_DATA_TYPE_UINT: case TSDB_DATA_TYPE_INT: @@ -1691,7 +1691,7 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) { int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { SVariant* pVal = &pCtx->param[1].param; - int32_t code = 0; + int32_t code = 0; double v = 0; GET_TYPED_DATA(v, double, pVal->nType, &pVal->i); @@ -2070,7 +2070,7 @@ static void prepareBuf(SqlFunctionCtx* pCtx) { } static int32_t firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SqlFunctionCtx* pCtx, - SFirstLastRes* pInfo) { + SFirstLastRes* pInfo) { int32_t code = TSDB_CODE_SUCCESS; if (pCtx->subsidiaries.num <= 0) { @@ -2123,7 +2123,8 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { } // All null data column, return directly. - if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && pInputCol->hasNull == true) { + if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && + pInputCol->hasNull == true) { // save selectivity value for column consisted of all null values int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo); if (code != TSDB_CODE_SUCCESS) { @@ -2239,7 +2240,8 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } // All null data column, return directly. - if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && pInputCol->hasNull == true) { + if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) && + pInputCol->hasNull == true) { // save selectivity value for column consisted of all null values int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo); if (code != TSDB_CODE_SUCCESS) { @@ -2333,7 +2335,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { } if (pResInfo->numOfRes == 0 || pInfo->ts < cts) { - char* data = colDataGetData(pInputCol, chosen); + char* data = colDataGetData(pInputCol, chosen); int32_t code = doSaveCurrentVal(pCtx, i, cts, type, data); if (code != TSDB_CODE_SUCCESS) { return code; @@ -2344,7 +2346,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { for (int32_t i = pInput->startRowIndex + round * 4; i < pInput->startRowIndex + pInput->numOfRows; ++i) { if (pResInfo->numOfRes == 0 || pInfo->ts < pts[i]) { - char* data = colDataGetData(pInputCol, i); + char* data = colDataGetData(pInputCol, i); int32_t code = doSaveCurrentVal(pCtx, i, pts[i], type, data); if (code != TSDB_CODE_SUCCESS) { return code; @@ -2361,7 +2363,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { numOfElems++; if (pResInfo->numOfRes == 0 || pInfo->ts < pts[i]) { - char* data = colDataGetData(pInputCol, i); + char* data = colDataGetData(pInputCol, i); int32_t code = doSaveCurrentVal(pCtx, i, pts[i], type, data); if (code != TSDB_CODE_SUCCESS) { return code; @@ -2408,7 +2410,7 @@ static int32_t firstLastTransferInfoImpl(SFirstLastRes* pInput, SFirstLastRes* p } static int32_t firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst, - int32_t rowIndex) { + int32_t rowIndex) { if (TSDB_CODE_SUCCESS == firstLastTransferInfoImpl(pInput, pOutput, isFirst)) { int32_t code = firstlastSaveTupleData(pCtx->pSrcBlock, rowIndex, pCtx, pOutput); if (code != TSDB_CODE_SUCCESS) { @@ -2435,7 +2437,7 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer for (int32_t i = start; i < start + pInput->numOfRows; ++i) { char* data = colDataGetData(pCol, i); SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data); - int32_t code = firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery, i); + int32_t code = firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery, i); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -2667,7 +2669,7 @@ static int32_t doSetPrevVal(SDiffInfo* pDiffInfo, int32_t type, const char* pv, } static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv, SColumnInfoData* pOutput, int32_t pos, - int32_t order, int64_t ts) { + int32_t order, int64_t ts) { int32_t factor = (order == TSDB_ORDER_ASC) ? 1 : -1; pDiffInfo->prevTs = ts; switch (type) { @@ -2875,8 +2877,8 @@ static STopBotRes* getTopBotOutputInfo(SqlFunctionCtx* pCtx) { return pRes; } -static int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type, - uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery); +static int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, + uint16_t type, uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery); static void addResult(SqlFunctionCtx* pCtx, STopBotResItem* pSourceItem, int16_t type, bool isTopQuery); @@ -2897,7 +2899,7 @@ int32_t topFunction(SqlFunctionCtx* pCtx) { } numOfElems++; - char* data = colDataGetData(pCol, i); + char* data = colDataGetData(pCol, i); int32_t code = doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, true); if (code != TSDB_CODE_SUCCESS) { return code; @@ -2931,7 +2933,7 @@ int32_t bottomFunction(SqlFunctionCtx* pCtx) { } numOfElems++; - char* data = colDataGetData(pCol, i); + char* data = colDataGetData(pCol, i); int32_t code = doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, false); if (code != TSDB_CODE_SUCCESS) { return code; @@ -2983,7 +2985,7 @@ static int32_t topBotResComparFn(const void* p1, const void* p2, const void* par } int32_t doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type, - uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery) { + uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery) { STopBotRes* pRes = getTopBotOutputInfo(pCtx); SVariant val = {0}; @@ -3174,7 +3176,7 @@ static char* doLoadTupleData(SSerializeDataHandle* pHandle, const STuplePos* pPo if (pPage == NULL) { return NULL; } - char* p = pPage->data + pPos->offset; + char* p = pPage->data + pPos->offset; releaseBufPage(pHandle->pBuf, pPage); return p; } else { @@ -4635,7 +4637,7 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) { continue; } - char* data = colDataGetData(pInputCol, i); + char* data = colDataGetData(pInputCol, i); int32_t code = doReservoirSample(pCtx, pInfo, data, i); if (code != TSDB_CODE_SUCCESS) { return code; @@ -4655,7 +4657,7 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) { } int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { - int32_t code = TSDB_CODE_SUCCESS; + int32_t code = TSDB_CODE_SUCCESS; SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx); SSampleInfo* pInfo = getSampleOutputInfo(pCtx); @@ -4989,7 +4991,7 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) { } numOfElems++; - char* data = colDataGetData(pInputCol, i); + char* data = colDataGetData(pInputCol, i); int32_t code = doModeAdd(pInfo, i, pCtx, data); if (code != TSDB_CODE_SUCCESS) { return code; @@ -5410,6 +5412,7 @@ int32_t blockDistFunction(SqlFunctionCtx* pCtx) { if (pDistInfo->maxRows < p1.maxRows) { pDistInfo->maxRows = p1.maxRows; } + pDistInfo->numOfVgroups += (p1.numOfTables != 0 ? 1 : 0); for (int32_t i = 0; i < tListLen(pDistInfo->blockRowsHisto); ++i) { pDistInfo->blockRowsHisto[i] += p1.blockRowsHisto[i]; @@ -5438,6 +5441,7 @@ int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDist if (tEncodeI32(&encoder, pInfo->defMinRows) < 0) return -1; if (tEncodeU32(&encoder, pInfo->numOfInmemRows) < 0) return -1; if (tEncodeU32(&encoder, pInfo->numOfSmallBlocks) < 0) return -1; + if (tEncodeU32(&encoder, pInfo->numOfVgroups) < 0) return -1; for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) { if (tEncodeI32(&encoder, pInfo->blockRowsHisto[i]) < 0) return -1; @@ -5469,6 +5473,7 @@ int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo if (tDecodeI32(&decoder, &pInfo->defMinRows) < 0) return -1; if (tDecodeU32(&decoder, &pInfo->numOfInmemRows) < 0) return -1; if (tDecodeU32(&decoder, &pInfo->numOfSmallBlocks) < 0) return -1; + if (tDecodeU32(&decoder, &pInfo->numOfVgroups) < 0) return -1; for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) { if (tDecodeI32(&decoder, &pInfo->blockRowsHisto[i]) < 0) return -1; @@ -5520,7 +5525,7 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { colDataAppend(pColInfo, row++, st, false); len = sprintf(st + VARSTR_HEADER_SIZE, "Total_Tables=[%d] Total_Files=[%d] Total_Vgroups=[%d]", pData->numOfTables, - pData->numOfFiles, 0); + pData->numOfFiles, pData->numOfVgroups); varDataSetLen(st, len); colDataAppend(pColInfo, row++, st, false); diff --git a/source/libs/qworker/inc/qwInt.h b/source/libs/qworker/inc/qwInt.h index f14df8e57c56ac237c4c93528b36a1535569ebbf..283fc7aa10884d408aa3daeaeeab9ff0007f46df 100644 --- a/source/libs/qworker/inc/qwInt.h +++ b/source/libs/qworker/inc/qwInt.h @@ -76,6 +76,7 @@ typedef struct SQWDebug { bool lockEnable; bool statusEnable; bool dumpEnable; + bool forceStop; bool sleepSimulate; bool deadSimulate; bool redirectSimulate; @@ -248,6 +249,7 @@ typedef struct SQWorkerMgmt { #define QW_QUERY_RUNNING(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY) #define QW_FETCH_RUNNING(ctx) ((ctx)->inFetch) +#define QW_QUERY_NOT_STARTED(ctx) (QW_GET_PHASE(ctx) == -1) #define QW_SET_QTID(id, qId, tId, eId) \ do { \ diff --git a/source/libs/qworker/src/qwDbg.c b/source/libs/qworker/src/qwDbg.c index 7a755cd36f46f87674096bae5b970a137c4e4655..0ab501ddd5abe806f62ef611900df1455b716e9f 100644 --- a/source/libs/qworker/src/qwDbg.c +++ b/source/libs/qworker/src/qwDbg.c @@ -9,11 +9,13 @@ #include "tmsg.h" #include "tname.h" -SQWDebug gQWDebug = {.statusEnable = true, +SQWDebug gQWDebug = {.lockEnable = false, + .statusEnable = true, .dumpEnable = false, .redirectSimulate = false, .deadSimulate = false, - .sleepSimulate = false}; + .sleepSimulate = false, + .forceStop = false}; int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore) { if (!gQWDebug.statusEnable) { @@ -306,6 +308,12 @@ int32_t qwDbgEnableDebug(char *option) { return TSDB_CODE_SUCCESS; } + if (0 == strcasecmp(option, "forceStop")) { + gQWDebug.forceStop = true; + qError("qw forceStop debug enabled"); + return TSDB_CODE_SUCCESS; + } + qError("invalid qw debug option:%s", option); return TSDB_CODE_APP_ERROR; diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index 2df0109d1912fa63b7c0719ac4579157939304ae..e2db0b8319e1db910c3e1d84e9c5663a772b9c6b 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -18,6 +18,51 @@ SQWorkerMgmt gQwMgmt = { .qwNum = 0, }; + +int32_t qwStopAllTasks(SQWorker *mgmt) { + uint64_t qId, tId, sId; + int32_t eId; + int64_t rId = 0; + + void *pIter = taosHashIterate(mgmt->ctxHash, NULL); + while (pIter) { + SQWTaskCtx *ctx = (SQWTaskCtx *)pIter; + void *key = taosHashGetKey(pIter, NULL); + QW_GET_QTID(key, qId, tId, eId); + + QW_LOCK(QW_WRITE, &ctx->lock); + + sId = ctx->sId; + + QW_TASK_DLOG_E("start to force stop task"); + + if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP) || QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) { + QW_TASK_WLOG_E("task already dropping"); + QW_UNLOCK(QW_WRITE, &ctx->lock); + + pIter = taosHashIterate(mgmt->ctxHash, pIter); + continue; + } + + if (QW_QUERY_RUNNING(ctx)) { + qwKillTaskHandle(ctx, TSDB_CODE_VND_STOPPED); + QW_TASK_DLOG_E("task running, async killed"); + } else if (QW_FETCH_RUNNING(ctx)) { + QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED); + QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP); + QW_TASK_DLOG_E("task fetching, update drop received"); + } else { + qwDropTask(QW_FPARAMS()); + } + + QW_UNLOCK(QW_WRITE, &ctx->lock); + + pIter = taosHashIterate(mgmt->ctxHash, pIter); + } + + return TSDB_CODE_SUCCESS; +} + int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t code = 0; SSchedulerHbRsp rsp = {0}; @@ -973,6 +1018,10 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { qwDbgDumpMgmtInfo(mgmt); + if (gQWDebug.forceStop) { + (void)qwStopAllTasks(mgmt); + } + QW_LOCK(QW_READ, &mgmt->schLock); int32_t schNum = taosHashGetSize(mgmt->schHash); @@ -1087,6 +1136,7 @@ _return: QW_RET(TSDB_CODE_SUCCESS); } + int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, void **qWorkerMgmt, const SMsgCb *pMsgCb) { if (NULL == qWorkerMgmt || (pMsgCb && pMsgCb->mgmt == NULL)) { qError("invalid param to init qworker"); @@ -1185,46 +1235,10 @@ void qWorkerStopAllTasks(void *qWorkerMgmt) { SQWorker *mgmt = (SQWorker *)qWorkerMgmt; QW_DLOG("start to stop all tasks, taskNum:%d", taosHashGetSize(mgmt->ctxHash)); - - uint64_t qId, tId, sId; - int32_t eId; - int64_t rId = 0; - + atomic_store_8(&mgmt->nodeStopped, 1); - void *pIter = taosHashIterate(mgmt->ctxHash, NULL); - while (pIter) { - SQWTaskCtx *ctx = (SQWTaskCtx *)pIter; - void *key = taosHashGetKey(pIter, NULL); - QW_GET_QTID(key, qId, tId, eId); - - QW_LOCK(QW_WRITE, &ctx->lock); - - sId = ctx->sId; - - QW_TASK_DLOG_E("start to force stop task"); - - if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP) || QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) { - QW_TASK_WLOG_E("task already dropping"); - QW_UNLOCK(QW_WRITE, &ctx->lock); - - pIter = taosHashIterate(mgmt->ctxHash, pIter); - continue; - } - - if (QW_QUERY_RUNNING(ctx)) { - qwKillTaskHandle(ctx, TSDB_CODE_VND_STOPPED); - } else if (QW_FETCH_RUNNING(ctx)) { - QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED); - QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP); - } else { - qwDropTask(QW_FPARAMS()); - } - - QW_UNLOCK(QW_WRITE, &ctx->lock); - - pIter = taosHashIterate(mgmt->ctxHash, pIter); - } + (void)qwStopAllTasks(mgmt); } void qWorkerDestroy(void **qWorkerMgmt) { diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 8faaf041083e1668bc1e7aa7c38d09bd2aef7671..829db74d3f204c5c440ee50e4f9d373ca8101258 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -727,7 +727,7 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) { QUEUE_INIT(&conn->q); conn->hostThrd = pThrd; conn->status = ConnNormal; - conn->broken = 0; + conn->broken = false; transRefCliHandle(conn); atomic_add_fetch_32(&pThrd->connCount, 1); @@ -997,6 +997,11 @@ static void cliDestroyBatch(SCliBatch* pBatch) { taosMemoryFree(pBatch); } static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) { + if (pThrd->quit == true) { + cliDestroyBatch(pBatch); + return; + } + if (pBatch == NULL || pBatch->wLen == 0 || QUEUE_IS_EMPTY(&pBatch->wq)) { return; } @@ -1082,17 +1087,23 @@ static void cliSendBatchCb(uv_write_t* req, int status) { if (status != 0) { tDebug("%s conn %p failed to send batch msg, batch size:%d, msgLen:%d, reason:%s", CONN_GET_INST_LABEL(conn), conn, p->wLen, p->batchSize, uv_err_name(status)); - cliHandleExcept(conn); + + if (!uv_is_closing((uv_handle_t*)&conn->stream)) cliHandleExcept(conn); + cliHandleBatchReq(nxtBatch, thrd); } else { tDebug("%s conn %p succ to send batch msg, batch size:%d, msgLen:%d", CONN_GET_INST_LABEL(conn), conn, p->wLen, p->batchSize); - - if (nxtBatch != NULL) { - conn->pBatch = nxtBatch; - cliSendBatch(conn); + if (!uv_is_closing((uv_handle_t*)&conn->stream)) { + if (nxtBatch != NULL) { + conn->pBatch = nxtBatch; + cliSendBatch(conn); + } else { + addConnToPool(thrd->pool, conn); + } } else { - addConnToPool(thrd->pool, conn); + cliDestroyBatch(nxtBatch); + // conn release by other callback } } @@ -1454,6 +1465,11 @@ static void cliNoBatchDealReq(queue* wq, SCliThrd* pThrd) { QUEUE_REMOVE(h); SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); + + if (pMsg->type == Quit) { + pThrd->stopMsg = pMsg; + continue; + } (*cliAsyncHandle[pMsg->type])(pMsg, pThrd); count++; @@ -1485,6 +1501,12 @@ static void cliBatchDealReq(queue* wq, SCliThrd* pThrd) { QUEUE_REMOVE(h); SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); + + if (pMsg->type == Quit) { + pThrd->stopMsg = pMsg; + continue; + } + if (pMsg->type == Normal && REQUEST_NO_RESP(&pMsg->msg)) { STransConnCtx* pCtx = pMsg->ctx; @@ -1582,7 +1604,6 @@ static void cliAsyncCb(uv_async_t* handle) { SCliThrd* pThrd = item->pThrd; STrans* pTransInst = pThrd->pTransInst; - SCliMsg* pMsg = NULL; // batch process to avoid to lock/unlock frequently queue wq; taosThreadMutexLock(&item->mtx); @@ -2285,22 +2306,8 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); return TSDB_CODE_RPC_BROKEN_LINK; } - /*if (pTransInst->connLimitNum > 0 && REQUEST_NO_RESP(pReq)) { - char key[TSDB_FQDN_LEN + 64] = {0}; - char* ip = EPSET_GET_INUSE_IP((SEpSet*)pEpSet); - uint16_t port = EPSET_GET_INUSE_PORT((SEpSet*)pEpSet); - CONN_CONSTRUCT_HASH_KEY(key, ip, port); - - int32_t* val = taosHashGet(pThrd->connLimitCache, key, strlen(key)); - if (val != NULL && *val >= pTransInst->connLimitNum) { - transFreeMsg(pReq->pCont); - transReleaseExHandle(transGetInstMgt(), (int64_t)shandle); - return TSDB_CODE_RPC_MAX_SESSIONS; - } - }*/ TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64()); - STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); pCtx->epSet = *pEpSet; pCtx->origEpSet = *pEpSet; diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index 5581931178486ba05f36640edcf7500477cbc96c..631bcb443ebecdc89ab3338d5dd5254964232093 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -22,7 +22,7 @@ typedef void *(*ThreadFp)(void *param); int32_t tQWorkerInit(SQWorkerPool *pool) { pool->qset = taosOpenQset(); - pool->workers = taosMemoryCalloc(pool->max, sizeof(SQWorker)); + pool->workers = taosMemoryCalloc(pool->max, sizeof(SQueueWorker)); if (pool->workers == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -31,7 +31,7 @@ int32_t tQWorkerInit(SQWorkerPool *pool) { (void)taosThreadMutexInit(&pool->mutex, NULL); for (int32_t i = 0; i < pool->max; ++i) { - SQWorker *worker = pool->workers + i; + SQueueWorker *worker = pool->workers + i; worker->id = i; worker->pool = pool; } @@ -42,14 +42,14 @@ int32_t tQWorkerInit(SQWorkerPool *pool) { void tQWorkerCleanup(SQWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { - SQWorker *worker = pool->workers + i; + SQueueWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { taosQsetThreadResume(pool->qset); } } for (int32_t i = 0; i < pool->max; ++i) { - SQWorker *worker = pool->workers + i; + SQueueWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { uInfo("worker:%s:%d is stopping", pool->name, worker->id); taosThreadJoin(worker->thread, NULL); @@ -65,7 +65,7 @@ void tQWorkerCleanup(SQWorkerPool *pool) { uInfo("worker:%s is closed", pool->name); } -static void *tQWorkerThreadFp(SQWorker *worker) { +static void *tQWorkerThreadFp(SQueueWorker *worker) { SQWorkerPool *pool = worker->pool; SQueueInfo qinfo = {0}; void *msg = NULL; @@ -106,7 +106,7 @@ STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) { // spawn a thread to process queue if (pool->num < pool->max) { do { - SQWorker *worker = pool->workers + pool->num; + SQueueWorker *worker = pool->workers + pool->num; TdThreadAttr thAttr; taosThreadAttrInit(&thAttr); @@ -138,7 +138,7 @@ void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) { int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) { pool->qset = taosOpenQset(); - pool->workers = taosArrayInit(2, sizeof(SQWorker *)); + pool->workers = taosArrayInit(2, sizeof(SQueueWorker *)); if (pool->workers == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -153,14 +153,14 @@ int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) { void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) { int32_t size = taosArrayGetSize(pool->workers); for (int32_t i = 0; i < size; ++i) { - SQWorker *worker = taosArrayGetP(pool->workers, i); + SQueueWorker *worker = taosArrayGetP(pool->workers, i); if (taosCheckPthreadValid(worker->thread)) { taosQsetThreadResume(pool->qset); } } for (int32_t i = 0; i < size; ++i) { - SQWorker *worker = taosArrayGetP(pool->workers, i); + SQueueWorker *worker = taosArrayGetP(pool->workers, i); if (taosCheckPthreadValid(worker->thread)) { uInfo("worker:%s:%d is stopping", pool->name, worker->id); taosThreadJoin(worker->thread, NULL); @@ -177,7 +177,7 @@ void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) { uInfo("worker:%s is closed", pool->name); } -static void *tAutoQWorkerThreadFp(SQWorker *worker) { +static void *tAutoQWorkerThreadFp(SQueueWorker *worker) { SAutoQWorkerPool *pool = worker->pool; SQueueInfo qinfo = {0}; void *msg = NULL; @@ -222,7 +222,7 @@ STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem // spawn a thread to process queue while (curWorkerNum < dstWorkerNum) { - SQWorker *worker = taosMemoryCalloc(1, sizeof(SQWorker)); + SQueueWorker *worker = taosMemoryCalloc(1, sizeof(SQueueWorker)); if (worker == NULL || taosArrayPush(pool->workers, &worker) == NULL) { uError("worker:%s:%d failed to create", pool->name, curWorkerNum); taosMemoryFree(worker);