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

fix client taskqueueWorkerpool stuck problem when querypolicy = 4

This commit is contained in:
wangjiaming0909 2024-09-09 10:02:47 +08:00
parent 9f70866ddb
commit b39b6589c4
9 changed files with 14 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/qcom/query.h
View File

@ -348,6 +348,8 @@ extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSi
void* (*mallocFp)(int64_t));
extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize);
void* getTaskPoolWorkerCb();
#define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE
#define SET_META_TYPE_CTABLE(t) (t) = META_TYPE_CTABLE
#define SET_META_TYPE_TABLE(t) (t) = META_TYPE_TABLE

1
include/libs/qworker/qworker.h
View File

@ -72,6 +72,7 @@ typedef struct SQWMsg {
int32_t msgLen;
SQWMsgInfo msgInfo;
SRpcHandleInfo connInfo;
void *pWorkerCb;
} SQWMsg;
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, void **qWorkerMgmt, const SMsgCb *pMsgCb);

1
include/libs/scheduler/scheduler.h
View File

@ -67,6 +67,7 @@ typedef struct SSchedulerReq {
SExecResult* pExecRes;
void** pFetchRes;
int8_t source;
void* pWorkerCb;
} SSchedulerReq;
int32_t schedulerInit(void);

2
source/client/src/clientImpl.c
View File

@ -841,6 +841,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
.chkKillParam = (void*)pRequest->self,
.pExecRes = &res,
.source = pRequest->source,
.pWorkerCb = getTaskPoolWorkerCb(),
};
int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob);
@ -1358,6 +1359,7 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat
.chkKillParam = (void*)pRequest->self,
.pExecRes = NULL,
.source = pRequest->source,
.pWorkerCb = getTaskPoolWorkerCb(),
};
if (TSDB_CODE_SUCCESS == code) {
code = schedulerExecJob(&req, &pRequest->body.queryJob);

4
source/libs/qcom/src/queryUtil.c
View File

@ -671,3 +671,7 @@ void freeDbCfgInfo(SDbCfgInfo* pInfo) {
}
taosMemoryFree(pInfo);
}
void* getTaskPoolWorkerCb() {
return taskQueue.wrokrerPool.pCb;
}

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

@ -1478,6 +1478,7 @@ int32_t qWorkerProcessLocalQuery(void *pMgmt, uint64_t sId, uint64_t qId, uint64
ctx->explainRes = explainRes;
rHandle.pMsgCb = taosMemoryCalloc(1, sizeof(SMsgCb));
rHandle.pWorkerCb = qwMsg->pWorkerCb;
if (NULL == rHandle.pMsgCb) {
QW_ERR_JRET(terrno);
}

1
source/libs/scheduler/inc/schInt.h
View File

@ -306,6 +306,7 @@ typedef struct SSchJob {
char *sql;
SQueryProfileSummary summary;
int8_t source;
void *pWorkerCb;
} SSchJob;
typedef struct SSchTaskCtx {

1
source/libs/scheduler/src/schJob.c
View File

@ -836,6 +836,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) {
pJob->userRes.execFp = pReq->execFp;
pJob->userRes.cbParam = pReq->cbParam;
pJob->source = pReq->source;
pJob->pWorkerCb = pReq->pWorkerCb;
if (pReq->pNodeList == NULL || taosArrayGetSize(pReq->pNodeList) <= 0) {
qDebug("qid:0x%" PRIx64 " input exec nodeList is empty", pReq->pDag->queryId);

1
source/libs/scheduler/src/schTask.c
View File

@ -1134,6 +1134,7 @@ int32_t schLaunchLocalTask(SSchJob *pJob, SSchTask *pTask) {
qwMsg.msg = pTask->plan;
qwMsg.msgType = pTask->plan->msgType;
qwMsg.connInfo.handle = pJob->conn.pTrans;
qwMsg.pWorkerCb = pJob->pWorkerCb;
if (SCH_IS_EXPLAIN_JOB(pJob)) {
explainRes = taosArrayInit(pJob->taskNum, sizeof(SExplainLocalRsp));