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:57:39 +08:00
parent 5ae72c15d4
commit 12d27bfdd3
4 changed files with 13 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/dnode/qnode/qnode.h
View File

@ -72,7 +72,6 @@ int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad);
* @param pMsg The request message * @param pMsg The request message
*/ */
int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg); int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg);
int32_t qndProcessFetchMsg(SQnode *pQnode, SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

6
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
View File

@ -35,12 +35,8 @@ static void qmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
case TDMT_MON_QM_INFO: case TDMT_MON_QM_INFO:
code = qmProcessGetMonitorInfoReq(pMgmt, pMsg); code = qmProcessGetMonitorInfoReq(pMgmt, pMsg);
break; break;
case TDMT_VND_QUERY:
case TDMT_VND_QUERY_CONTINUE:
code = qndProcessQueryMsg(pMgmt->pQnode, pMsg);
break;
default: default:
code = qndProcessFetchMsg(pMgmt->pQnode, pMsg); code = qndProcessQueryMsg(pMgmt->pQnode, pMsg);
break; break;
} }

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

@ -18,34 +18,19 @@
#include "mndMnode.h" #include "mndMnode.h"
#include "qworker.h" #include "qworker.h"
int32_t mndProcessQueryMsg(SRpcMsg *pReq) { int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
int32_t code = -1; int32_t code = -1;
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pMsg->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", pMsg);
switch (pReq->msgType) { switch (pMsg->msgType) {
case TDMT_VND_QUERY: case TDMT_VND_QUERY:
code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pReq); code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pMsg);
break; break;
case TDMT_VND_QUERY_CONTINUE: case TDMT_VND_QUERY_CONTINUE:
code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pReq); code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pMsg);
break; break;
default:
terrno = TSDB_CODE_VND_APP_ERROR;
mError("unknown msg type:%d in query queue", pReq->msgType);
}
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
return code;
}
int32_t mndProcessFetchMsg(SRpcMsg *pMsg) {
int32_t code = -1;
SMnode *pMnode = pMsg->info.node;
mTrace("msg:%p, in fetch queue is processing", pMsg);
switch (pMsg->msgType) {
case TDMT_VND_FETCH: case TDMT_VND_FETCH:
code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg); code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg);
break; break;
@ -57,7 +42,7 @@ int32_t mndProcessFetchMsg(SRpcMsg *pMsg) {
break; break;
default: default:
terrno = TSDB_CODE_VND_APP_ERROR; terrno = TSDB_CODE_VND_APP_ERROR;
mError("unknown msg type:%d in fetch queue", pMsg->msgType); mError("unknown msg type:%d in query queue", pMsg->msgType);
} }
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS; if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
@ -72,9 +57,9 @@ int32_t mndInitQuery(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_VND_QUERY, mndProcessQueryMsg); mndSetMsgHandle(pMnode, TDMT_VND_QUERY, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_VND_QUERY_CONTINUE, mndProcessQueryMsg); mndSetMsgHandle(pMnode, TDMT_VND_QUERY_CONTINUE, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_VND_FETCH, mndProcessFetchMsg); mndSetMsgHandle(pMnode, TDMT_VND_FETCH, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_VND_DROP_TASK, mndProcessFetchMsg); mndSetMsgHandle(pMnode, TDMT_VND_DROP_TASK, mndProcessQueryMsg);
mndSetMsgHandle(pMnode, TDMT_VND_QUERY_HEARTBEAT, mndProcessFetchMsg); mndSetMsgHandle(pMnode, TDMT_VND_QUERY_HEARTBEAT, mndProcessQueryMsg);
return 0; return 0;
} }

18
source/dnode/qnode/src/qnode.c
View File

@ -45,7 +45,7 @@ int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) { return 0; }
int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) { int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
int32_t code = -1; int32_t code = -1;
SReadHandle handle = {.pMsgCb = &pQnode->msgCb}; SReadHandle handle = {.pMsgCb = &pQnode->msgCb};
qTrace("message in qnode query queue is processing"); qTrace("message in qnode queue is processing");
switch (pMsg->msgType) { switch (pMsg->msgType) {
case TDMT_VND_QUERY: case TDMT_VND_QUERY:
@ -54,20 +54,6 @@ int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
case TDMT_VND_QUERY_CONTINUE: case TDMT_VND_QUERY_CONTINUE:
code = qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg); code = qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg);
break; break;
default:
qError("unknown msg type:%d in query queue", pMsg->msgType);
terrno = TSDB_CODE_VND_APP_ERROR;
}
if (code == 0) return TSDB_CODE_ACTION_IN_PROGRESS;
return code;
}
int32_t qndProcessFetchMsg(SQnode *pQnode, SRpcMsg *pMsg) {
int32_t code = -1;
qTrace("message in fetch queue is processing");
switch (pMsg->msgType) {
case TDMT_VND_FETCH: case TDMT_VND_FETCH:
code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg); code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg);
break; break;
@ -96,7 +82,7 @@ int32_t qndProcessFetchMsg(SQnode *pQnode, SRpcMsg *pMsg) {
code = qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg); code = qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg);
break; break;
default: default:
qError("unknown msg type:%d in fetch queue", pMsg->msgType); qError("unknown msg type:%d in qnode queue", pMsg->msgType);
terrno = TSDB_CODE_VND_APP_ERROR; terrno = TSDB_CODE_VND_APP_ERROR;
} }