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

refactor: rpc msg handler

This commit is contained in:
Shengliang Guan 2022-05-21 17:38:09 +08:00
parent 1146196fd3
commit d5fabd5375
2 changed files with 21 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
View File

@ -56,24 +56,6 @@ static void mmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static void mmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SMnodeMgmt *pMgmt = pInfo->ahandle;
int32_t code = -1;
dTrace("msg:%p, get from mnode-query queue", pMsg);
pMsg->info.node = pMgmt->pMnode;
code = mndProcessMsg(pMsg);
if (IsReq(pMsg) && pMsg->info.handle != NULL && code != 0) {
if (terrno != 0) code = terrno;
mmSendRsp(pMsg, code);
}
dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
rpcFreeCont(pMsg->pCont);
taosFreeQitem(pMsg);
}
static int32_t mmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) { static int32_t mmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) {
dTrace("msg:%p, put into worker %s, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType)); dTrace("msg:%p, put into worker %s, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType));
taosWriteQitem(pWorker->queue, pMsg); taosWriteQitem(pWorker->queue, pMsg);
@ -131,7 +113,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = tsNumOfMnodeQueryThreads, .min = tsNumOfMnodeQueryThreads,
.max = tsNumOfMnodeQueryThreads, .max = tsNumOfMnodeQueryThreads,
.name = "mnode-query", .name = "mnode-query",
.fp = (FItem)mmProcessQueryQueue, .fp = (FItem)mmProcessQueue,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->queryWorker, &qCfg) != 0) { if (tSingleWorkerInit(&pMgmt->queryWorker, &qCfg) != 0) {

27
source/dnode/mnode/impl/src/mndQuery.c
View File

@ -19,36 +19,49 @@
#include "qworker.h" #include "qworker.h"
int32_t mndProcessQueryMsg(SRpcMsg *pReq) { int32_t mndProcessQueryMsg(SRpcMsg *pReq) {
int32_t code = -1;
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb}; SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb};
mTrace("msg:%p, in query queue is processing", pReq); mTrace("msg:%p, in query queue is processing", pReq);
switch (pReq->msgType) { switch (pReq->msgType) {
case TDMT_VND_QUERY: case TDMT_VND_QUERY:
return qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pReq); code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pReq);
break;
case TDMT_VND_QUERY_CONTINUE: case TDMT_VND_QUERY_CONTINUE:
return qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pReq); code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pReq);
break;
default: default:
terrno = TSDB_CODE_VND_APP_ERROR;
mError("unknown msg type:%d in query queue", pReq->msgType); mError("unknown msg type:%d in query queue", pReq->msgType);
return TSDB_CODE_VND_APP_ERROR;
} }
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
return code;
} }
int32_t mndProcessFetchMsg(SRpcMsg *pMsg) { int32_t mndProcessFetchMsg(SRpcMsg *pMsg) {
int32_t code = -1;
SMnode *pMnode = pMsg->info.node; SMnode *pMnode = pMsg->info.node;
mTrace("msg:%p, in fetch queue is processing", pMsg); mTrace("msg:%p, in fetch queue is processing", pMsg);
switch (pMsg->msgType) { switch (pMsg->msgType) {
case TDMT_VND_FETCH: case TDMT_VND_FETCH:
return qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg); code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg);
break;
case TDMT_VND_DROP_TASK: case TDMT_VND_DROP_TASK:
return qWorkerProcessDropMsg(pMnode, pMnode->pQuery, pMsg); code = qWorkerProcessDropMsg(pMnode, pMnode->pQuery, pMsg);
break;
case TDMT_VND_QUERY_HEARTBEAT: case TDMT_VND_QUERY_HEARTBEAT:
return qWorkerProcessHbMsg(pMnode, pMnode->pQuery, pMsg); code = qWorkerProcessHbMsg(pMnode, pMnode->pQuery, pMsg);
break;
default: default:
terrno = TSDB_CODE_VND_APP_ERROR;
mError("unknown msg type:%d in fetch queue", pMsg->msgType); mError("unknown msg type:%d in fetch queue", pMsg->msgType);
return TSDB_CODE_VND_APP_ERROR;
} }
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
return code;
} }
int32_t mndInitQuery(SMnode *pMnode) { int32_t mndInitQuery(SMnode *pMnode) {