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

fix: error handling issue

This commit is contained in:
dapan1121 2024-07-15 13:50:04 +08:00
parent 4c2c9a71d5
commit a82b0b805d
3 changed files with 32 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/qworker/src/qwDbg.c
View File

@ -260,8 +260,8 @@ void qwDbgSimulateDead(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *rsped) {
if (++ignoreTime > 10 && 0 == taosRand() % 9) {
if (ctx->fetchMsgType == TDMT_SCH_FETCH) {
qwBuildAndSendErrorRsp(TDMT_SCH_LINK_BROKEN, &ctx->ctrlConnInfo, TSDB_CODE_RPC_BROKEN_LINK);
qwBuildAndSendErrorRsp(ctx->fetchMsgType + 1, &ctx->dataConnInfo, TSDB_CODE_QRY_TASK_CTX_NOT_EXIST);
(void)qwBuildAndSendErrorRsp(TDMT_SCH_LINK_BROKEN, &ctx->ctrlConnInfo, TSDB_CODE_RPC_BROKEN_LINK); // ignore error
(void)qwBuildAndSendErrorRsp(ctx->fetchMsgType + 1, &ctx->dataConnInfo, TSDB_CODE_QRY_TASK_CTX_NOT_EXIST); // ignore error
*rsped = true;
taosSsleep(3);

27
source/libs/qworker/src/qwMsg.c
View File

@ -105,8 +105,19 @@ int32_t qwBuildAndSendExplainRsp(SRpcHandleInfo *pConn, SArray *pExecList) {
SExplainRsp rsp = {.numOfPlans = taosArrayGetSize(pExecList), .subplanInfo = pInfo};
int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp);
if (contLen < 0) {
qError("tSerializeSExplainRsp failed, error: %x", terrno);
QW_RET(terrno);
}
void *pRsp = rpcMallocCont(contLen);
tSerializeSExplainRsp(pRsp, contLen, &rsp);
if (NULL == pRsp) {
QW_RET(terrno);
}
contLen = tSerializeSExplainRsp(pRsp, contLen, &rsp);
if (contLen < 0) {
qError("tSerializeSExplainRsp second failed, error: %x", terrno);
QW_RET(terrno);
}
SRpcMsg rpcRsp = {
.msgType = TDMT_SCH_EXPLAIN_RSP,
@ -123,8 +134,20 @@ int32_t qwBuildAndSendExplainRsp(SRpcHandleInfo *pConn, SArray *pExecList) {
int32_t qwBuildAndSendHbRsp(SRpcHandleInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) {
int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus);
if (contLen < 0) {
qError("tSerializeSSchedulerHbRsp failed, error: %x", terrno);
QW_RET(terrno);
}
void *pRsp = rpcMallocCont(contLen);
tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus);
if (NULL == pRsp) {
QW_RET(terrno);
}
contLen = tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus);
if (contLen < 0) {
qError("tSerializeSSchedulerHbRsp second failed, error: %x", terrno);
QW_RET(terrno);
}
SRpcMsg rpcRsp = {
.msgType = TDMT_SCH_QUERY_HEARTBEAT_RSP,

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

@ -682,7 +682,10 @@ _return:
qwDbgSimulateRedirect(&qwMsg, ctx, &rsped);
qwDbgSimulateDead(QW_FPARAMS(), ctx, &rsped);
if (!rsped) {
code = qwSendQueryRsp(QW_FPARAMS(), input->msgType + 1, ctx, code, false);
int32_t newCode = qwSendQueryRsp(QW_FPARAMS(), input->msgType + 1, ctx, code, false);
if (TSDB_CODE_SUCCESS != newCode && TSDB_CODE_SUCCESS == code) {
code = newCode;
}
}
}
@ -694,7 +697,7 @@ _return:
}
if (code) {
code = qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_FAIL, ctx->dynamicTask);
(void)qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_FAIL, ctx->dynamicTask); // already in error, ignore new error
}
QW_UNLOCK(QW_WRITE, &ctx->lock);