xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: drop task while task executing issue

This commit is contained in:
dapan1121 2023-01-03 18:07:18 +08:00
parent 5660199016
commit 1477ce97a7
3 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/executor/executor.h
View File

@ -154,6 +154,8 @@ void qCleanExecTaskBlockBuf(qTaskInfo_t tinfo);
*/
int32_t qAsyncKillTask(qTaskInfo_t tinfo, int32_t rspCode);
bool qTaskIsExecuting(qTaskInfo_t qinfo);
/**
* destroy query info structure
* @param qHandle

9
source/libs/executor/src/executor.c
View File

@ -704,6 +704,15 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {
return TSDB_CODE_SUCCESS;
}
bool qTaskIsExecuting(qTaskInfo_t qinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
if (NULL == pTaskInfo) {
return false;
}
return 0 != atomic_load_64(&pTaskInfo->owner);
}
static void printTaskExecCostInLog(SExecTaskInfo* pTaskInfo) {
STaskCostInfo* pSummary = &pTaskInfo->cost;
int64_t idleTime = pSummary->start - pSummary->created;

15
source/libs/qworker/src/qworker.c
View File

@ -202,6 +202,15 @@ _return:
QW_RET(code);
}
bool qwTaskNotInExec(SQWTaskCtx *ctx) {
qTaskInfo_t taskHandle = ctx->taskHandle;
if (NULL == taskHandle || !qTaskIsExecuting(taskHandle)) {
return true;
}
return false;
}
int32_t qwGenerateSchHbRsp(SQWorker *mgmt, SQWSchStatus *sch, SQWHbInfo *hbInfo) {
int32_t taskNum = 0;
@ -508,8 +517,10 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp
}
if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
QW_ERR_JRET(ctx->rspCode);
if (QW_PHASE_POST_FETCH != phase || qwTaskNotInExec(ctx)) {
QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
QW_ERR_JRET(ctx->rspCode);
}
}
if (ctx->rspCode) {