diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 253563448042cd0f9153389e9b3f65c715ddcaae..0cb58528676e37bdf5b7fbaf54e37d87f93c5950 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -1718,8 +1718,7 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) { pTableMetaInfo->pTableMeta = (STableMeta *) taosCachePut(tscMetaCache, pTableMetaInfo->name, strlen(pTableMetaInfo->name), pTableMeta, size, tsTableMetaKeepTimer * 1000); - - // todo handle out of memory case + if (pTableMetaInfo->pTableMeta == NULL) { free(pTableMeta); return TSDB_CODE_TSC_OUT_OF_MEMORY; diff --git a/src/inc/tsdb.h b/src/inc/tsdb.h index fe032777637cae37040a171179cae5caddc8531a..02ca99e4b85f5dfedb8b868803f2135c36eb01df 100644 --- a/src/inc/tsdb.h +++ b/src/inc/tsdb.h @@ -164,6 +164,12 @@ typedef struct STsdbQueryCond { SColumnInfo *colList; } STsdbQueryCond; +typedef struct SMemRef { + int32_t ref; + void *mem; + void *imem; +} SMemRef; + typedef struct SDataBlockInfo { STimeWindow window; int32_t rows; @@ -193,7 +199,7 @@ typedef struct { * @param qinfo query info handle from query processor * @return */ -TsdbQueryHandleT *tsdbQueryTables(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, void *qinfo); +TsdbQueryHandleT *tsdbQueryTables(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, void *qinfo, SMemRef* pRef); /** * Get the last row of the given query time window for all the tables in STableGroupInfo object. @@ -205,7 +211,7 @@ TsdbQueryHandleT *tsdbQueryTables(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STab * @param tableInfo table list. * @return */ -TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, void *qinfo); +TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, void *qinfo, SMemRef* pRef); /** * get the queried table object list @@ -223,7 +229,7 @@ SArray* tsdbGetQueriedTableList(TsdbQueryHandleT *pHandle); * @return */ TsdbQueryHandleT tsdbQueryRowsInExternalWindow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, - void *qinfo); + void *qinfo, SMemRef* pRef); /** * move to next block if exists diff --git a/src/query/inc/qExecutor.h b/src/query/inc/qExecutor.h index d7db897cdf9973019536a7b9bc8f8a4d7a8bc85d..22397d0314c3081e8ed23d9671f4ccfcb7abc244 100644 --- a/src/query/inc/qExecutor.h +++ b/src/query/inc/qExecutor.h @@ -200,12 +200,6 @@ enum { QUERY_RESULT_READY = 2, }; -typedef struct SMemRef { - int32_t ref; - void *mem; - void *imem; -} SMemRef; - typedef struct SQInfo { void* signature; int32_t code; // error code to returned to client diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index b992a037ed6f37bb674a0d7755653abe491ed845..10417f03b81984b66a9d38d80410e52292e78203 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -1711,6 +1711,19 @@ _clean: return TSDB_CODE_QRY_OUT_OF_MEMORY; } +static void doFreeQueryHandle(SQInfo* pQInfo) { + SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv; + + tsdbCleanupQueryHandle(pRuntimeEnv->pQueryHandle); + tsdbCleanupQueryHandle(pRuntimeEnv->pSecQueryHandle); + + pRuntimeEnv->pQueryHandle = NULL; + pRuntimeEnv->pSecQueryHandle = NULL; + + SMemRef* pMemRef = &pQInfo->memRef; + assert(pMemRef->ref == 0 && pMemRef->imem == NULL && pMemRef->mem == NULL); +} + static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) { if (pRuntimeEnv->pQuery == NULL) { return; @@ -1740,8 +1753,7 @@ static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) { pRuntimeEnv->pFillInfo = taosDestroyFillInfo(pRuntimeEnv->pFillInfo); destroyResultBuf(pRuntimeEnv->pResultBuf); - tsdbCleanupQueryHandle(pRuntimeEnv->pQueryHandle); - tsdbCleanupQueryHandle(pRuntimeEnv->pSecQueryHandle); + doFreeQueryHandle(pQInfo); pRuntimeEnv->pTSBuf = tsBufDestroy(pRuntimeEnv->pTSBuf); taosTFree(pRuntimeEnv->keyBuf); @@ -3536,7 +3548,7 @@ static void setEnvBeforeReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusI tsdbCleanupQueryHandle(pRuntimeEnv->pSecQueryHandle); } - pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); if (pRuntimeEnv->pSecQueryHandle == NULL) { longjmp(pRuntimeEnv->env, terrno); } @@ -3620,7 +3632,7 @@ void scanOneTableDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) { } restoreTimeWindow(&pQInfo->tableGroupInfo, &cond); - pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); if (pRuntimeEnv->pSecQueryHandle == NULL) { longjmp(pRuntimeEnv->env, terrno); } @@ -4462,7 +4474,7 @@ static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) terrno = TSDB_CODE_SUCCESS; if (isFirstLastRowQuery(pQuery)) { - pRuntimeEnv->pQueryHandle = tsdbQueryLastRow(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryLastRow(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); // update the query time window pQuery->window = cond.twindow; @@ -4484,9 +4496,9 @@ static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) } } } else if (isPointInterpoQuery(pQuery)) { - pRuntimeEnv->pQueryHandle = tsdbQueryRowsInExternalWindow(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryRowsInExternalWindow(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); } else { - pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); } return terrno; @@ -4765,7 +4777,7 @@ static bool multiTableMultioutputHelper(SQInfo *pQInfo, int32_t index) { pRuntimeEnv->pQueryHandle = NULL; } - pRuntimeEnv->pQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &gp, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &gp, pQInfo, &pQInfo->memRef); taosArrayDestroy(tx); taosArrayDestroy(g1); if (pRuntimeEnv->pQueryHandle == NULL) { @@ -4880,7 +4892,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) { pRuntimeEnv->pQueryHandle = NULL; } - pRuntimeEnv->pQueryHandle = tsdbQueryRowsInExternalWindow(pQInfo->tsdb, &cond, &gp, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryRowsInExternalWindow(pQInfo->tsdb, &cond, &gp, pQInfo, &pQInfo->memRef); taosArrayDestroy(tx); taosArrayDestroy(g1); @@ -4946,7 +4958,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) { } // no need to update the lastkey for each table - pRuntimeEnv->pQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &gp, pQInfo); + pRuntimeEnv->pQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &gp, pQInfo, &pQInfo->memRef); taosArrayDestroy(g1); taosArrayDestroy(tx); @@ -5155,7 +5167,7 @@ static void doSaveContext(SQInfo *pQInfo) { setupQueryRangeForReverseScan(pQInfo); pRuntimeEnv->prevGroupId = INT32_MIN; - pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo); + pRuntimeEnv->pSecQueryHandle = tsdbQueryTables(pQInfo->tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo, &pQInfo->memRef); if (pRuntimeEnv->pSecQueryHandle == NULL) { longjmp(pRuntimeEnv->env, terrno); } @@ -6841,7 +6853,7 @@ int32_t qRetrieveQueryResultInfo(qinfo_t qinfo, bool* buildRes, void* pRspContex int32_t code = TSDB_CODE_SUCCESS; -#if 0 +#if _NON_BLOCKING_RETRIEVE SQuery *pQuery = pQInfo->runtimeEnv.pQuery; pthread_mutex_lock(&pQInfo->lock); @@ -6913,6 +6925,8 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co pQInfo->dataReady = QUERY_RESULT_NOT_READY; if (IS_QUERY_KILLED(pQInfo) || Q_STATUS_EQUAL(pQuery->status, QUERY_OVER)) { + // here current thread hold the refcount, so it is safe to free tsdbQueryHandle. + doFreeQueryHandle(pQInfo); *continueExec = false; (*pRsp)->completed = 1; // notify no more result to client } else { diff --git a/src/tsdb/src/tsdbMemTable.c b/src/tsdb/src/tsdbMemTable.c index d68a97eb137ad04f123f122f665dc4931f16c8d5..150bda3b800b8598fad9bcdfaba571f448ddd0ea 100644 --- a/src/tsdb/src/tsdbMemTable.c +++ b/src/tsdb/src/tsdbMemTable.c @@ -198,6 +198,8 @@ void tsdbUnTakeMemSnapShot(STsdbRepo *pRepo, SMemTable *pMem, SMemTable *pIMem) if (pIMem != NULL) { tsdbUnRefMemTable(pRepo, pIMem); } + + tsdbDebug("vgId:%d utake memory snapshot, pMem %p pIMem %p", REPO_ID(pRepo), pMem, pIMem); } void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) { diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index 25fff48c49c794565cb727514759dfc60d9191aa..27f49b19024ee199e980f38d49646e44a40b7fd3 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -20,7 +20,6 @@ #include "exception.h" #include "../../query/inc/qAst.h" // todo move to common module -#include "../../query/inc/qExecutor.h" // todo move to common module #include "tlosertree.h" #include "tsdb.h" #include "tsdbMain.h" @@ -120,8 +119,9 @@ typedef struct STsdbQueryHandle { SDataCols *pDataCols; // in order to hold current file data block int32_t allocSize; // allocated data block size - SMemTable *mem; // mem-table - SMemTable *imem; // imem-table, acquired from snapshot + SMemRef *pMemRef; +// SMemTable *mem; // mem-table +// SMemTable *imem; // imem-table, acquired from snapshot SArray *defaultLoadColumn;// default load column SDataBlockLoadInfo dataBlockLoadInfo; /* record current block load information */ SLoadCompBlockInfo compBlockLoadInfo; /* record current compblock information in SQuery */ @@ -184,26 +184,26 @@ static SArray* getDefaultLoadColumns(STsdbQueryHandle* pQueryHandle, bool loadTS return pLocalIdList; } -static void tsdbMayTakeMemSnapshot(TsdbQueryHandleT pHandle) { - STsdbQueryHandle* pSecQueryHandle = (STsdbQueryHandle*) pHandle; - SQInfo *pQInfo = (SQInfo *)(pSecQueryHandle->qinfo); +static void tsdbMayTakeMemSnapshot(STsdbQueryHandle* pQueryHandle) { + assert(pQueryHandle != NULL && pQueryHandle->pMemRef != NULL); - if (pQInfo->memRef.ref++ == 0) { - tsdbTakeMemSnapshot(pSecQueryHandle->pTsdb, &pSecQueryHandle->mem, &pSecQueryHandle->imem); - pQInfo->memRef.mem = pSecQueryHandle->mem; - pQInfo->memRef.imem = pSecQueryHandle->imem; - } else { - pSecQueryHandle->mem = (SMemTable *)(pQInfo->memRef.mem); - pSecQueryHandle->imem = (SMemTable *)(pQInfo->memRef.imem); + SMemRef* pMemRef = pQueryHandle->pMemRef; + if (pQueryHandle->pMemRef->ref++ == 0) { + tsdbTakeMemSnapshot(pQueryHandle->pTsdb, (SMemTable**)&(pMemRef->mem), (SMemTable**)&(pMemRef->imem)); } } -static void tsdbMayUnTakeMemSnapshot(TsdbQueryHandleT pHandle) { - STsdbQueryHandle* pSecQueryHandle = (STsdbQueryHandle*) pHandle; - SQInfo *pQInfo = (SQInfo *)(pSecQueryHandle->qinfo); - if (--pQInfo->memRef.ref == 0) { - tsdbUnTakeMemSnapShot(pSecQueryHandle->pTsdb, pSecQueryHandle->mem, pSecQueryHandle->imem); +static void tsdbMayUnTakeMemSnapshot(STsdbQueryHandle* pQueryHandle) { + assert(pQueryHandle != NULL && pQueryHandle->pMemRef != NULL); + + SMemRef* pMemRef = pQueryHandle->pMemRef; + if (--pMemRef->ref == 0) { + tsdbUnTakeMemSnapShot(pQueryHandle->pTsdb, pMemRef->mem, pMemRef->imem); + pMemRef->mem = NULL; + pMemRef->imem = NULL; } + + pQueryHandle->pMemRef = NULL; } static SArray* createCheckInfoFromTableGroup(STsdbQueryHandle* pQueryHandle, STableGroupInfo* pGroupList, STsdbMeta* pMeta) { size_t sizeOfGroup = taosArrayGetSize(pGroupList->pGroupList); @@ -270,7 +270,7 @@ static SArray* createCheckInfoFromCheckInfo(SArray* pTableCheckInfo, TSKEY skey) return pNew; } -static STsdbQueryHandle* tsdbQueryTablesImpl(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, void* qinfo) { +static STsdbQueryHandle* tsdbQueryTablesImpl(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, void* qinfo, SMemRef* pMemRef) { STsdbQueryHandle* pQueryHandle = calloc(1, sizeof(STsdbQueryHandle)); if (pQueryHandle == NULL) { goto out_of_memory; @@ -288,13 +288,14 @@ static STsdbQueryHandle* tsdbQueryTablesImpl(TSDB_REPO_T* tsdb, STsdbQueryCond* pQueryHandle->outputCapacity = ((STsdbRepo*)tsdb)->config.maxRowsPerFileBlock; pQueryHandle->allocSize = 0; pQueryHandle->locateStart = false; + pQueryHandle->pMemRef = pMemRef; if (tsdbInitReadHelper(&pQueryHandle->rhelper, (STsdbRepo*) tsdb) != 0) { goto out_of_memory; } tsdbMayTakeMemSnapshot(pQueryHandle); - assert(pCond != NULL && pCond->numOfCols > 0); + assert(pCond != NULL && pCond->numOfCols > 0 && pMemRef != NULL); if (ASCENDING_TRAVERSE(pCond->order)) { assert(pQueryHandle->window.skey <= pQueryHandle->window.ekey); @@ -348,8 +349,8 @@ static STsdbQueryHandle* tsdbQueryTablesImpl(TSDB_REPO_T* tsdb, STsdbQueryCond* return NULL; } -TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList, void* qinfo) { - STsdbQueryHandle* pQueryHandle = tsdbQueryTablesImpl(tsdb, pCond, qinfo); +TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList, void* qinfo, SMemRef* pRef) { + STsdbQueryHandle* pQueryHandle = tsdbQueryTablesImpl(tsdb, pCond, qinfo, pRef); STsdbMeta* pMeta = tsdbGetMeta(tsdb); assert(pMeta != NULL); @@ -366,7 +367,7 @@ TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STab return (TsdbQueryHandleT) pQueryHandle; } -TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, void* qinfo) { +TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, void* qinfo, SMemRef* pMemRef) { pCond->twindow = changeTableGroupByLastrow(groupList); // no qualified table @@ -374,7 +375,7 @@ TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STab return NULL; } - STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qinfo); + STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qinfo, pMemRef); assert(pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey <= pCond->twindow.ekey); return pQueryHandle; @@ -396,8 +397,8 @@ SArray* tsdbGetQueriedTableList(TsdbQueryHandleT *pHandle) { return res; } -TsdbQueryHandleT tsdbQueryRowsInExternalWindow(TSDB_REPO_T *tsdb, STsdbQueryCond* pCond, STableGroupInfo *groupList, void* qinfo) { - STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qinfo); +TsdbQueryHandleT tsdbQueryRowsInExternalWindow(TSDB_REPO_T *tsdb, STsdbQueryCond* pCond, STableGroupInfo *groupList, void* qinfo, SMemRef* pRef) { + STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qinfo, pRef); if (pQueryHandle != NULL) { pQueryHandle->type = TSDB_QUERY_TYPE_EXTERNAL; changeQueryHandleForInterpQuery(pQueryHandle); @@ -417,7 +418,7 @@ static bool initTableMemIterator(STsdbQueryHandle* pHandle, STableCheckInfo* pCh int32_t order = pHandle->order; // no data in buffer, abort - if (pHandle->mem == NULL && pHandle->imem == NULL) { + if (pHandle->pMemRef->mem == NULL && pHandle->pMemRef->imem == NULL) { return false; } @@ -426,16 +427,19 @@ static bool initTableMemIterator(STsdbQueryHandle* pHandle, STableCheckInfo* pCh STableData* pMem = NULL; STableData* pIMem = NULL; - if (pHandle->mem && pCheckInfo->tableId.tid < pHandle->mem->maxTables) { - pMem = pHandle->mem->tData[pCheckInfo->tableId.tid]; + SMemTable* pMemT = pHandle->pMemRef->mem; + SMemTable* pIMemT = pHandle->pMemRef->imem; + + if (pMemT && pCheckInfo->tableId.tid < pMemT->maxTables) { + pMem = pMemT->tData[pCheckInfo->tableId.tid]; if (pMem != NULL && pMem->uid == pCheckInfo->tableId.uid) { // check uid pCheckInfo->iter = tSkipListCreateIterFromVal(pMem->pData, (const char*)&pCheckInfo->lastKey, TSDB_DATA_TYPE_TIMESTAMP, order); } } - if (pHandle->imem && pCheckInfo->tableId.tid < pHandle->imem->maxTables) { - pIMem = pHandle->imem->tData[pCheckInfo->tableId.tid]; + if (pIMemT && pCheckInfo->tableId.tid < pIMemT->maxTables) { + pIMem = pIMemT->tData[pCheckInfo->tableId.tid]; if (pIMem != NULL && pIMem->uid == pCheckInfo->tableId.uid) { // check uid pCheckInfo->iiter = tSkipListCreateIterFromVal(pIMem->pData, (const char*)&pCheckInfo->lastKey, TSDB_DATA_TYPE_TIMESTAMP, order); @@ -2029,7 +2033,7 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) { memcpy(&cond.colList[i], &pColInfoData->info, sizeof(SColumnInfo)); } - STsdbQueryHandle* pSecQueryHandle = tsdbQueryTablesImpl(pQueryHandle->pTsdb, &cond, pQueryHandle->qinfo); + STsdbQueryHandle* pSecQueryHandle = tsdbQueryTablesImpl(pQueryHandle->pTsdb, &cond, pQueryHandle->qinfo, pQueryHandle->pMemRef); taosTFree(cond.colList);