提交 52951dc9 编写于 作者: H Haojun Liao

[td-225] fix error in qinfo life-cycle management.

上级 1b4fa8da
...@@ -44,7 +44,7 @@ void qDestroyQueryInfo(qinfo_t qinfo); ...@@ -44,7 +44,7 @@ void qDestroyQueryInfo(qinfo_t qinfo);
* @param qinfo * @param qinfo
* @return * @return
*/ */
void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param); void qTableQuery(qinfo_t qinfo);
/** /**
* Retrieve the produced results information, if current query is not paused or completed, * Retrieve the produced results information, if current query is not paused or completed,
......
...@@ -197,7 +197,8 @@ typedef struct SQInfo { ...@@ -197,7 +197,8 @@ typedef struct SQInfo {
*/ */
int32_t tableIndex; int32_t tableIndex;
int32_t numOfGroupResultPages; int32_t numOfGroupResultPages;
_qinfo_free_fn_t fn; _qinfo_free_fn_t freeFn;
jmp_buf env;
} SQInfo; } SQInfo;
#endif // TDENGINE_QUERYEXECUTOR_H #endif // TDENGINE_QUERYEXECUTOR_H
...@@ -2586,7 +2586,6 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) { ...@@ -2586,7 +2586,6 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
} }
int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) { int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) {
// int64_t maxOutput = 0;
for (int32_t j = 0; j < pQuery->numOfOutput; ++j) { for (int32_t j = 0; j < pQuery->numOfOutput; ++j) {
int32_t functionId = pQuery->pSelectExpr[j].base.functionId; int32_t functionId = pQuery->pSelectExpr[j].base.functionId;
...@@ -2604,15 +2603,6 @@ int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) { ...@@ -2604,15 +2603,6 @@ int64_t getNumOfResultWindowRes(SQuery *pQuery, SWindowResult *pWindowRes) {
if (pResultInfo->numOfRes > 0) { if (pResultInfo->numOfRes > 0) {
return pResultInfo->numOfRes; return pResultInfo->numOfRes;
} }
// if (pResultInfo != NULL && maxOutput < pResultInfo->numOfRes) {
// maxOutput = pResultInfo->numOfRes;
//
// if (maxOutput > 0) {
// break;
// }
// }
//
// assert(pResultInfo != NULL);
} }
return 0; return 0;
...@@ -2623,12 +2613,19 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) { ...@@ -2623,12 +2613,19 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
size_t size = taosArrayGetSize(pGroup); size_t size = taosArrayGetSize(pGroup);
tFilePage **buffer = pQuery->sdata; tFilePage **buffer = pQuery->sdata;
int32_t * posList = calloc(size, sizeof(int32_t));
int32_t* posList = calloc(size, sizeof(int32_t));
STableQueryInfo **pTableList = malloc(POINTER_BYTES * size); STableQueryInfo **pTableList = malloc(POINTER_BYTES * size);
if (pTableList == NULL || posList == NULL) {
tfree(posList);
tfree(pTableList);
qError("QInfo:%p failed alloc memory", pQInfo);
longjmp(pQInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
// todo opt for the case of one table per group // todo opt for the case of one table per group
int32_t numOfTables = 0; int32_t numOfTables = 0;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
...@@ -4069,7 +4066,7 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) { ...@@ -4069,7 +4066,7 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
return pFillCol; return pFillCol;
} }
int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bool isSTableQuery, void* freeParam, _qinfo_free_fn_t fn) { int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bool isSTableQuery) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
...@@ -4083,8 +4080,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo ...@@ -4083,8 +4080,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
pQInfo->tsdb = tsdb; pQInfo->tsdb = tsdb;
pQInfo->vgId = vgId; pQInfo->vgId = vgId;
pQInfo->param = freeParam;
pQInfo->fn = fn;
pRuntimeEnv->pQuery = pQuery; pRuntimeEnv->pQuery = pQuery;
pRuntimeEnv->pTSBuf = pTsBuf; pRuntimeEnv->pTSBuf = pTsBuf;
...@@ -4932,14 +4927,6 @@ static void tableQueryImpl(SQInfo *pQInfo) { ...@@ -4932,14 +4927,6 @@ static void tableQueryImpl(SQInfo *pQInfo) {
// record the total elapsed time // record the total elapsed time
pRuntimeEnv->summary.elapsedTime += (taosGetTimestampUs() - st); pRuntimeEnv->summary.elapsedTime += (taosGetTimestampUs() - st);
assert(pQInfo->tableqinfoGroupInfo.numOfTables == 1); assert(pQInfo->tableqinfoGroupInfo.numOfTables == 1);
/* check if query is killed or not */
if (isQueryKilled(pQInfo)) {
qDebug("QInfo:%p query is killed", pQInfo);
} else {
qDebug("QInfo:%p query paused, %" PRId64 " rows returned, numOfTotal:%" PRId64 " rows",
pQInfo, pQuery->rec.rows, pQuery->rec.total + pQuery->rec.rows);
}
} }
static void stableQueryImpl(SQInfo *pQInfo) { static void stableQueryImpl(SQInfo *pQInfo) {
...@@ -4961,10 +4948,6 @@ static void stableQueryImpl(SQInfo *pQInfo) { ...@@ -4961,10 +4948,6 @@ static void stableQueryImpl(SQInfo *pQInfo) {
// record the total elapsed time // record the total elapsed time
pQInfo->runtimeEnv.summary.elapsedTime += (taosGetTimestampUs() - st); pQInfo->runtimeEnv.summary.elapsedTime += (taosGetTimestampUs() - st);
if (pQuery->rec.rows == 0) {
qDebug("QInfo:%p over, %zu tables queried, %"PRId64" rows are returned", pQInfo, pQInfo->tableqinfoGroupInfo.numOfTables, pQuery->rec.total);
}
} }
static int32_t getColumnIndexInSource(SQueryTableMsg *pQueryMsg, SSqlFuncMsg *pExprMsg, SColumnInfo* pTagCols) { static int32_t getColumnIndexInSource(SQueryTableMsg *pQueryMsg, SSqlFuncMsg *pExprMsg, SColumnInfo* pTagCols) {
...@@ -5076,6 +5059,8 @@ static char *createTableIdList(SQueryTableMsg *pQueryMsg, char *pMsg, SArray **p ...@@ -5076,6 +5059,8 @@ static char *createTableIdList(SQueryTableMsg *pQueryMsg, char *pMsg, SArray **p
*/ */
static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, SSqlFuncMsg ***pExpr, static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, SSqlFuncMsg ***pExpr,
char **tagCond, char** tbnameCond, SColIndex **groupbyCols, SColumnInfo** tagCols) { char **tagCond, char** tbnameCond, SColIndex **groupbyCols, SColumnInfo** tagCols) {
int32_t code = TSDB_CODE_SUCCESS;
pQueryMsg->numOfTables = htonl(pQueryMsg->numOfTables); pQueryMsg->numOfTables = htonl(pQueryMsg->numOfTables);
pQueryMsg->window.skey = htobe64(pQueryMsg->window.skey); pQueryMsg->window.skey = htobe64(pQueryMsg->window.skey);
...@@ -5102,7 +5087,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5102,7 +5087,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
// query msg safety check // query msg safety check
if (!validateQueryMsg(pQueryMsg)) { if (!validateQueryMsg(pQueryMsg)) {
return TSDB_CODE_QRY_INVALID_MSG; code = TSDB_CODE_QRY_INVALID_MSG;
goto _cleanup;
} }
char *pMsg = (char *)(pQueryMsg->colList) + sizeof(SColumnInfo) * pQueryMsg->numOfCols; char *pMsg = (char *)(pQueryMsg->colList) + sizeof(SColumnInfo) * pQueryMsg->numOfCols;
...@@ -5174,7 +5160,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5174,7 +5160,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
int16_t functionId = pExprMsg->functionId; int16_t functionId = pExprMsg->functionId;
if (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAGPRJ || functionId == TSDB_FUNC_TAG_DUMMY) { if (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAGPRJ || functionId == TSDB_FUNC_TAG_DUMMY) {
if (pExprMsg->colInfo.flag != TSDB_COL_TAG) { // ignore the column index check for arithmetic expression. if (pExprMsg->colInfo.flag != TSDB_COL_TAG) { // ignore the column index check for arithmetic expression.
return TSDB_CODE_QRY_INVALID_MSG; code = TSDB_CODE_QRY_INVALID_MSG;
goto _cleanup;
} }
} else { } else {
// if (!validateExprColumnInfo(pQueryMsg, pExprMsg)) { // if (!validateExprColumnInfo(pQueryMsg, pExprMsg)) {
...@@ -5186,6 +5173,7 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5186,6 +5173,7 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
} }
if (!validateQuerySourceCols(pQueryMsg, *pExpr)) { if (!validateQuerySourceCols(pQueryMsg, *pExpr)) {
code = TSDB_CODE_QRY_INVALID_MSG;
goto _cleanup; goto _cleanup;
} }
...@@ -5193,6 +5181,10 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5193,6 +5181,10 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
if (pQueryMsg->numOfGroupCols > 0) { // group by tag columns if (pQueryMsg->numOfGroupCols > 0) { // group by tag columns
*groupbyCols = malloc(pQueryMsg->numOfGroupCols * sizeof(SColIndex)); *groupbyCols = malloc(pQueryMsg->numOfGroupCols * sizeof(SColIndex));
if (*groupbyCols == NULL) {
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _cleanup;
}
for (int32_t i = 0; i < pQueryMsg->numOfGroupCols; ++i) { for (int32_t i = 0; i < pQueryMsg->numOfGroupCols; ++i) {
(*groupbyCols)[i].colId = *(int16_t *)pMsg; (*groupbyCols)[i].colId = *(int16_t *)pMsg;
...@@ -5248,7 +5240,13 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5248,7 +5240,13 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
if (*pMsg != 0) { if (*pMsg != 0) {
size_t len = strlen(pMsg) + 1; size_t len = strlen(pMsg) + 1;
*tbnameCond = malloc(len); *tbnameCond = malloc(len);
if (*tbnameCond == NULL) {
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _cleanup;
}
strcpy(*tbnameCond, pMsg); strcpy(*tbnameCond, pMsg);
pMsg += len; pMsg += len;
} }
...@@ -5258,7 +5256,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList, ...@@ -5258,7 +5256,8 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->queryType, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols, pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->queryType, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols,
pQueryMsg->order, pQueryMsg->numOfOutput, pQueryMsg->numOfCols, pQueryMsg->intervalTime, pQueryMsg->order, pQueryMsg->numOfOutput, pQueryMsg->numOfCols, pQueryMsg->intervalTime,
pQueryMsg->fillType, pQueryMsg->tsLen, pQueryMsg->tsNumOfBlocks, pQueryMsg->limit, pQueryMsg->offset); pQueryMsg->fillType, pQueryMsg->tsLen, pQueryMsg->tsNumOfBlocks, pQueryMsg->limit, pQueryMsg->offset);
return 0;
return TSDB_CODE_SUCCESS;
_cleanup: _cleanup:
tfree(*pExpr); tfree(*pExpr);
...@@ -5268,7 +5267,8 @@ _cleanup: ...@@ -5268,7 +5267,8 @@ _cleanup:
tfree(*groupbyCols); tfree(*groupbyCols);
tfree(*tagCols); tfree(*tagCols);
tfree(*tagCond); tfree(*tagCond);
return TSDB_CODE_QRY_INVALID_MSG;
return code;
} }
static int32_t buildAirthmeticExprFromMsg(SExprInfo *pArithExprInfo, SQueryTableMsg *pQueryMsg) { static int32_t buildAirthmeticExprFromMsg(SExprInfo *pArithExprInfo, SQueryTableMsg *pQueryMsg) {
...@@ -5669,7 +5669,8 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, ...@@ -5669,7 +5669,8 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
pQuery->window = pQueryMsg->window; pQuery->window = pQueryMsg->window;
if (sem_init(&pQInfo->dataReady, 0, 0) != 0) { if (sem_init(&pQInfo->dataReady, 0, 0) != 0) {
qError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, strerror(errno)); int32_t code = TAOS_SYSTEM_ERROR(errno);
qError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, tstrerror(code));
goto _cleanup; goto _cleanup;
} }
...@@ -5680,7 +5681,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, ...@@ -5680,7 +5681,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
_cleanup: _cleanup:
freeQInfo(pQInfo); freeQInfo(pQInfo);
return NULL; return NULL;
} }
...@@ -5722,6 +5722,9 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ ...@@ -5722,6 +5722,9 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
pQInfo->param = param;
pQInfo->freeFn = fn;
if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) { if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) {
qDebug("QInfo:%p no table qualified for tag filter, abort query", pQInfo); qDebug("QInfo:%p no table qualified for tag filter, abort query", pQInfo);
setQueryStatus(pQuery, QUERY_COMPLETED); setQueryStatus(pQuery, QUERY_COMPLETED);
...@@ -5731,7 +5734,7 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ ...@@ -5731,7 +5734,7 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
} }
// filter the qualified // filter the qualified
if ((code = doInitQInfo(pQInfo, pTSBuf, tsdb, vgId, isSTable, param, fn)) != TSDB_CODE_SUCCESS) { if ((code = doInitQInfo(pQInfo, pTSBuf, tsdb, vgId, isSTable)) != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
...@@ -6031,19 +6034,19 @@ void qDestroyQueryInfo(qinfo_t qHandle) { ...@@ -6031,19 +6034,19 @@ void qDestroyQueryInfo(qinfo_t qHandle) {
qDebug("QInfo:%p dec refCount, value:%d", pQInfo, ref); qDebug("QInfo:%p dec refCount, value:%d", pQInfo, ref);
if (ref == 0) { if (ref == 0) {
_qinfo_free_fn_t fn = pQInfo->fn; _qinfo_free_fn_t freeFp = pQInfo->freeFn;
void* param = pQInfo->param; void* param = pQInfo->param;
doDestoryQueryInfo(pQInfo); doDestoryQueryInfo(pQInfo);
if (fn != NULL) { if (freeFp != NULL) {
assert(param != NULL); assert(param != NULL);
fn(param); freeFp(param);
} }
} }
} }
void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) { void qTableQuery(qinfo_t qinfo) {
SQInfo *pQInfo = (SQInfo *)qinfo; SQInfo *pQInfo = (SQInfo *)qinfo;
if (pQInfo == NULL || pQInfo->signature != pQInfo) { if (pQInfo == NULL || pQInfo->signature != pQInfo) {
...@@ -6053,17 +6056,34 @@ void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) { ...@@ -6053,17 +6056,34 @@ void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) {
if (isQueryKilled(pQInfo)) { if (isQueryKilled(pQInfo)) {
qDebug("QInfo:%p it is already killed, abort", pQInfo); qDebug("QInfo:%p it is already killed, abort", pQInfo);
sem_post(&pQInfo->dataReady);
qDestroyQueryInfo(pQInfo); qDestroyQueryInfo(pQInfo);
return; return;
} }
if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) { if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) {
qDebug("QInfo:%p no table exists for query, abort", pQInfo); qDebug("QInfo:%p no table exists for query, abort", pQInfo);
sem_post(&pQInfo->dataReady);
qDestroyQueryInfo(pQInfo);
return;
}
int32_t ret = setjmp(pQInfo->env);
// error occurs, record the error code and return to client
if (ret != TSDB_CODE_SUCCESS) {
pQInfo->code = ret;
qDebug("QInfo:%p query abort due to error occurs, code:%s", pQInfo, tstrerror(pQInfo->code));
sem_post(&pQInfo->dataReady);
qDestroyQueryInfo(pQInfo);
return; return;
} }
qDebug("QInfo:%p query task is launched", pQInfo); qDebug("QInfo:%p query task is launched", pQInfo);
SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
if (onlyQueryTags(pQInfo->runtimeEnv.pQuery)) { if (onlyQueryTags(pQInfo->runtimeEnv.pQuery)) {
assert(pQInfo->runtimeEnv.pQueryHandle == NULL); assert(pQInfo->runtimeEnv.pQueryHandle == NULL);
buildTagQueryResult(pQInfo); // todo support the limit/offset buildTagQueryResult(pQInfo); // todo support the limit/offset
...@@ -6073,6 +6093,16 @@ void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) { ...@@ -6073,6 +6093,16 @@ void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) {
tableQueryImpl(pQInfo); tableQueryImpl(pQInfo);
} }
SQuery* pQuery = pRuntimeEnv->pQuery;
if (isQueryKilled(pQInfo)) {
qDebug("QInfo:%p query is killed", pQInfo);
} else if (pQuery->rec.rows == 0) {
qDebug("QInfo:%p over, %zu tables queried, %"PRId64" rows are returned", pQInfo, pQInfo->tableqinfoGroupInfo.numOfTables, pQuery->rec.total);
} else {
qDebug("QInfo:%p query paused, %" PRId64 " rows returned, numOfTotal:%" PRId64 " rows",
pQInfo, pQuery->rec.rows, pQuery->rec.total + pQuery->rec.rows);
}
sem_post(&pQInfo->dataReady); sem_post(&pQInfo->dataReady);
qDestroyQueryInfo(pQInfo); qDestroyQueryInfo(pQInfo);
} }
......
...@@ -1173,9 +1173,7 @@ tExprNode* exprTreeFromTableName(const char* tbnameCond) { ...@@ -1173,9 +1173,7 @@ tExprNode* exprTreeFromTableName(const char* tbnameCond) {
size_t len = strlen(cond) + VARSTR_HEADER_SIZE; size_t len = strlen(cond) + VARSTR_HEADER_SIZE;
char* p = exception_malloc(len); char* p = exception_malloc(len);
varDataSetLen(p, len - VARSTR_HEADER_SIZE); STR_WITH_SIZE_TO_VARSTR(p, cond, len - VARSTR_HEADER_SIZE);
memcpy(varDataVal(p), cond, len);
taosArrayPush(pVal->arr, &p); taosArrayPush(pVal->arr, &p);
} }
......
...@@ -547,7 +547,6 @@ void taosRemoveFromTrashCan(SCacheObj *pCacheObj, STrashElem *pElem) { ...@@ -547,7 +547,6 @@ void taosRemoveFromTrashCan(SCacheObj *pCacheObj, STrashElem *pElem) {
pCacheObj->freeFp(pElem->pData->data); pCacheObj->freeFp(pElem->pData->data);
} }
uError("-------------------free obj:%p", pElem->pData);
free(pElem->pData); free(pElem->pData);
free(pElem); free(pElem);
} }
......
...@@ -324,7 +324,7 @@ void vnodeRelease(void *pVnodeRaw) { ...@@ -324,7 +324,7 @@ void vnodeRelease(void *pVnodeRaw) {
assert(refCount >= 0); assert(refCount >= 0);
if (refCount > 0) { if (refCount > 0) {
vTrace("vgId:%d, release vnode, refCount:%d", vgId, refCount); vDebug("vgId:%d, release vnode, refCount:%d", vgId, refCount);
return; return;
} }
...@@ -388,7 +388,7 @@ void *vnodeAccquireVnode(int32_t vgId) { ...@@ -388,7 +388,7 @@ void *vnodeAccquireVnode(int32_t vgId) {
if (pVnode == NULL) return pVnode; if (pVnode == NULL) return pVnode;
atomic_add_fetch_32(&pVnode->refCount, 1); atomic_add_fetch_32(&pVnode->refCount, 1);
vTrace("vgId:%d, get vnode, refCount:%d", pVnode->vgId, pVnode->refCount); vDebug("vgId:%d, get vnode, refCount:%d", pVnode->vgId, pVnode->refCount);
return pVnode; return pVnode;
} }
...@@ -466,6 +466,7 @@ static void vnodeCleanUp(SVnodeObj *pVnode) { ...@@ -466,6 +466,7 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
vTrace("vgId:%d, vnode will cleanup, refCount:%d", pVnode->vgId, pVnode->refCount); vTrace("vgId:%d, vnode will cleanup, refCount:%d", pVnode->vgId, pVnode->refCount);
// release local resources only after cutting off outside connections // release local resources only after cutting off outside connections
taosCacheCleanup(pVnode->qHandlePool);
vnodeRelease(pVnode); vnodeRelease(pVnode);
} }
......
...@@ -78,15 +78,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { ...@@ -78,15 +78,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// this message arrived here by means of the *query* message, so release the vnode is necessary // this message arrived here by means of the *query* message, so release the vnode is necessary
void** qhandle = taosCacheAcquireByKey(pVnode->qHandlePool, (void*) &killQueryMsg->qhandle, sizeof(killQueryMsg->qhandle)); void** qhandle = taosCacheAcquireByKey(pVnode->qHandlePool, (void*) &killQueryMsg->qhandle, sizeof(killQueryMsg->qhandle));
if (qhandle == NULL || *qhandle == NULL) { // todo handle invalid qhandle error if (qhandle == NULL || *qhandle == NULL) {
vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void*) killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
} else { } else {
// qKillQuery((qinfo_t) killQueryMsg->qhandle);
taosCacheRelease(pVnode->qHandlePool, (void**) &qhandle, true); taosCacheRelease(pVnode->qHandlePool, (void**) &qhandle, true);
} }
vnodeRelease(pVnode); vnodeRelease(pVnode);
return TSDB_CODE_TSC_QUERY_CANCELLED; // todo change the error code return TSDB_CODE_TSC_QUERY_CANCELLED;
} }
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
...@@ -97,8 +96,8 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { ...@@ -97,8 +96,8 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
code = qCreateQueryInfo(pVnode->tsdb, pVnode->vgId, pQueryTableMsg, pVnode, vnodeRelease, &pQInfo); code = qCreateQueryInfo(pVnode->tsdb, pVnode->vgId, pQueryTableMsg, pVnode, vnodeRelease, &pQInfo);
SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp)); SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
pRsp->qhandle = htobe64((uint64_t) (pQInfo)); pRsp->code = code;
pRsp->code = code; pRsp->qhandle = 0;
pRet->len = sizeof(SQueryTableRsp); pRet->len = sizeof(SQueryTableRsp);
pRet->rsp = pRsp; pRet->rsp = pRsp;
...@@ -120,6 +119,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { ...@@ -120,6 +119,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
handle = taosCachePut(pVnode->qHandlePool, pQInfo, sizeof(pQInfo), &pQInfo, sizeof(pQInfo), tsShellActivityTimer * 2); handle = taosCachePut(pVnode->qHandlePool, pQInfo, sizeof(pQInfo), &pQInfo, sizeof(pQInfo), tsShellActivityTimer * 2);
assert(*handle == pQInfo); assert(*handle == pQInfo);
pRsp->qhandle = htobe64((uint64_t) (handle));
} else { } else {
assert(pQInfo == NULL); assert(pQInfo == NULL);
vnodeRelease(pVnode); vnodeRelease(pVnode);
...@@ -128,13 +128,17 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { ...@@ -128,13 +128,17 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
vDebug("vgId:%d, QInfo:%p, dnode query msg disposed", pVnode->vgId, pQInfo); vDebug("vgId:%d, QInfo:%p, dnode query msg disposed", pVnode->vgId, pQInfo);
} else { } else {
assert(pCont != NULL); assert(pCont != NULL);
pQInfo = pCont; pQInfo = *(void**)(pCont);
handle = pCont;
code = TSDB_CODE_VND_ACTION_IN_PROGRESS; code = TSDB_CODE_VND_ACTION_IN_PROGRESS;
vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, pQInfo); vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, pQInfo);
} }
if (pQInfo != NULL) { if (pQInfo != NULL) {
qTableQuery(pQInfo, vnodeRelease, pVnode); // do execute query qTableQuery(pQInfo); // do execute query
assert(handle != NULL);
taosCacheRelease(pVnode->qHandlePool, (void**) &handle, false); taosCacheRelease(pVnode->qHandlePool, (void**) &handle, false);
} }
...@@ -146,56 +150,57 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { ...@@ -146,56 +150,57 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
SRspRet *pRet = &pReadMsg->rspRet; SRspRet *pRet = &pReadMsg->rspRet;
SRetrieveTableMsg *pRetrieve = pCont; SRetrieveTableMsg *pRetrieve = pCont;
void *pQInfo = (void*) htobe64(pRetrieve->qhandle); void **pQInfo = (void*) htobe64(pRetrieve->qhandle);
pRetrieve->free = htons(pRetrieve->free); pRetrieve->free = htons(pRetrieve->free);
vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed", pVnode->vgId, *pQInfo);
memset(pRet, 0, sizeof(SRspRet)); memset(pRet, 0, sizeof(SRspRet));
int32_t ret = 0; int32_t ret = 0;
void** handle = taosCacheAcquireByKey(pVnode->qHandlePool, &pQInfo, sizeof(pQInfo)); void** handle = taosCacheAcquireByKey(pVnode->qHandlePool, pQInfo, sizeof(pQInfo));
if (handle == NULL || *handle != pQInfo) { if (handle == NULL || handle != pQInfo) {
ret = TSDB_CODE_QRY_INVALID_QHANDLE; ret = TSDB_CODE_QRY_INVALID_QHANDLE;
} }
if (pRetrieve->free == 1) { if (pRetrieve->free == 1) {
vDebug("vgId:%d, QInfo:%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, pQInfo); if (ret == TSDB_CODE_SUCCESS) {
vDebug("vgId:%d, QInfo:%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, pQInfo);
taosCacheRelease(pVnode->qHandlePool, handle, true);
// int32_t ret = qKillQuery(pQInfo);
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); taosCacheRelease(pVnode->qHandlePool, (void**) &handle, true);
pRet->len = sizeof(SRetrieveTableRsp); pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
pRet->len = sizeof(SRetrieveTableRsp);
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp)); memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
SRetrieveTableRsp* pRsp = pRet->rsp; SRetrieveTableRsp* pRsp = pRet->rsp;
pRsp->numOfRows = 0; pRsp->numOfRows = 0;
pRsp->completed = true; pRsp->completed = true;
pRsp->useconds = 0; pRsp->useconds = 0;
} else { // todo handle error
}
return ret; return ret;
} }
vDebug("vgId:%d, QInfo:%p, retrieve msg is received", pVnode->vgId, pQInfo); vDebug("vgId:%d, QInfo:%p, retrieve msg is received", pVnode->vgId, *pQInfo);
int32_t code = qRetrieveQueryResultInfo(pQInfo); int32_t code = qRetrieveQueryResultInfo(*pQInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
//TODO //TODO
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp)); memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
} else { } else {
// todo check code and handle error in build result set // todo check code and handle error in build result set
code = qDumpRetrieveResult(pQInfo, (SRetrieveTableRsp **)&pRet->rsp, &pRet->len); code = qDumpRetrieveResult(*pQInfo, (SRetrieveTableRsp **)&pRet->rsp, &pRet->len);
if (qHasMoreResultsToRetrieve(pQInfo)) { if (qHasMoreResultsToRetrieve(*pQInfo)) {
pRet->qhandle = pQInfo; pRet->qhandle = handle;
code = TSDB_CODE_VND_ACTION_NEED_REPROCESSED; code = TSDB_CODE_VND_ACTION_NEED_REPROCESSED;
} else { // no further execution invoked, release the ref to vnode } else { // no further execution invoked, release the ref to vnode
taosCacheRelease(pVnode->qHandlePool, (void**) &handle, true); taosCacheRelease(pVnode->qHandlePool, (void**) &handle, true);
// qDestroyQueryInfo(pQInfo);
} }
} }
vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed", pVnode->vgId, pQInfo);
return code; return code;
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册