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

Merge branch '3.0' into enh/opt-transport

This commit is contained in:
yihaoDeng 2024-09-16 18:05:30 +08:00
parent cf23956f4f
commit ef0c301a6a
1 changed files with 23 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
source/libs/transport/src/transCli.c
View File

@ -1507,8 +1507,7 @@ FORCE_INLINE int32_t cliBuildExceptResp(SCliThrd* pThrd, SCliReq* pReq, STransMs
// handle noresp and inter manage msg // handle noresp and inter manage msg
if (pCtx == NULL || REQUEST_NO_RESP(&pReq->msg)) { if (pCtx == NULL || REQUEST_NO_RESP(&pReq->msg)) {
destroyReq(pReq); return TSDB_CODE_RPC_NO_STATE;
return 0;
} }
if (pResp->code == 0) { if (pResp->code == 0) {
pResp->code = TSDB_CODE_RPC_BROKEN_LINK; pResp->code = TSDB_CODE_RPC_BROKEN_LINK;
@ -1613,22 +1612,30 @@ int32_t cliMayGetStateByQid(SCliThrd* pThrd, SCliReq* pReq, SCliConn** pConn) {
int64_t qid = pReq->msg.info.qId; int64_t qid = pReq->msg.info.qId;
if (qid == 0) { if (qid == 0) {
return TSDB_CODE_RPC_NO_STATE; return TSDB_CODE_RPC_NO_STATE;
} else {
SExHandle* exh = transAcquireExHandle(transGetRefMgt(), qid);
if (exh == NULL) {
return TSDB_CODE_RPC_NO_STATE;
} }
SReqState* pState = taosHashGet(pThrd->pIdConnTable, &qid, sizeof(qid)); SReqState* pState = taosHashGet(pThrd->pIdConnTable, &qid, sizeof(qid));
if (pState == NULL) { if (pState == NULL) {
if (pReq->ctx == NULL) { if (pReq->ctx == NULL) {
transReleaseExHandle(transGetRefMgt(), qid);
return TSDB_CODE_RPC_STATE_DROPED; return TSDB_CODE_RPC_STATE_DROPED;
} }
tDebug("%s conn %p failed to get statue, qid:%" PRId64 "", transLabel(pThrd->pInst), pConn, qid); tDebug("%s conn %p failed to get statue, qid:%" PRId64 "", transLabel(pThrd->pInst), pConn, qid);
transReleaseExHandle(transGetRefMgt(), qid);
return TSDB_CODE_RPC_ASYNC_IN_PROCESS; return TSDB_CODE_RPC_ASYNC_IN_PROCESS;
} else { } else {
*pConn = pState->conn; *pConn = pState->conn;
tDebug("%s conn %p succ to get conn of statue, qid:%" PRId64 "", transLabel(pThrd->pInst), pConn, qid); tDebug("%s conn %p succ to get conn of statue, qid:%" PRId64 "", transLabel(pThrd->pInst), pConn, qid);
} }
transReleaseExHandle(transGetRefMgt(), qid);
return 0; return 0;
} }
}
int32_t cliHandleState_mayUpdateState(SCliThrd* pThrd, SCliReq* pReq, SCliConn* pConn) { int32_t cliHandleState_mayUpdateState(SCliThrd* pThrd, SCliReq* pReq, SCliConn* pConn) {
int32_t code = 0; int32_t code = 0;
@ -1990,6 +1997,7 @@ int32_t notifyExceptCb(void* thrd, SCliReq* pReq, STransMsg* pResp) {
STrans* pInst = pThrd->pInst; STrans* pInst = pThrd->pInst;
int32_t code = cliBuildExceptResp(pThrd, pReq, pResp); int32_t code = cliBuildExceptResp(pThrd, pReq, pResp);
if (code != 0) { if (code != 0) {
destroyReq(pReq);
return code; return code;
} }
pInst->cfp(pInst->parent, pResp, NULL); pInst->cfp(pInst->parent, pResp, NULL);