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

Merge pull request #19372 from taosdata/FIX/TD-21546-main 

fix: send rpc response on closing sync or failing to enqueue
This commit is contained in:
Shengliang Guan 2023-01-05 09:13:17 +08:00 committed by GitHub
parent 56ab3a2315 442902e5cb
commit eb7b16a457
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -134,6 +134,13 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
}
}
static void vmSendResponse(SRpcMsg *pMsg) {
if (pMsg->info.handle) {
SRpcMsg rsp = {.info = pMsg->info, .code = terrno};
rpcSendResponse(&rsp);
}
}
static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {
const STraceId *trace = &pMsg->info.traceId;
if (pMsg->contLen < sizeof(SMsgHead)) {
@ -152,7 +159,9 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
if (pVnode == NULL) {
dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,
terrstr(), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
return terrno != 0 ? terrno : -1;
terrno = (terrno != 0) ? terrno : -1;
vmSendResponse(pMsg);
return terrno;
}
switch (qtype) {

1
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -182,7 +182,6 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
vError("vgId:%d, duplicate write request. version: %" PRId64 ", applied: %" PRId64 "", TD_VID(pVnode), version,
pVnode->state.applied);
terrno = TSDB_CODE_VND_DUP_REQUEST;
pRsp->info.handle = NULL;
return -1;
}

9
source/libs/sync/src/syncMain.c
View File

@ -1252,6 +1252,9 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
// stop heartbeat timer
syncNodeStopHeartbeatTimer(pSyncNode);
// clean rsp
syncRespCleanRsp(pSyncNode->pSyncRespMgr);
}
void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); }
@ -2667,16 +2670,12 @@ int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIn
}
int32_t code = syncNodeAppend(ths, pEntry);
if (code < 0) {
sNError(ths, "failed to append blocking msg");
}
return code;
} else {
syncEntryDestroy(pEntry);
pEntry = NULL;
return -1;
}
return -1;
}
int32_t syncNodeOnClientRequestOld(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex) {