From f4c3141821ec4d23e64ca4993d95ae561741ce11 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sat, 21 May 2022 18:10:37 +0800 Subject: [PATCH] refactor: rpc msg handler --- source/dnode/mgmt/mgmt_bnode/src/bmWorker.c | 8 +++---- source/dnode/mgmt/mgmt_dnode/src/dmWorker.c | 10 +++----- source/dnode/mgmt/mgmt_snode/src/smWorker.c | 8 +++---- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 23 +++++++++---------- source/dnode/mgmt/node_mgmt/inc/dmMgmt.h | 1 - source/dnode/mgmt/node_mgmt/src/dmMgmt.c | 5 ---- source/dnode/mgmt/node_mgmt/src/dmTransport.c | 3 ++- source/dnode/vnode/src/vnd/vnodeSvr.c | 3 --- 8 files changed, 22 insertions(+), 39 deletions(-) diff --git a/source/dnode/mgmt/mgmt_bnode/src/bmWorker.c b/source/dnode/mgmt/mgmt_bnode/src/bmWorker.c index e019924268..08c9edd854 100644 --- a/source/dnode/mgmt/mgmt_bnode/src/bmWorker.c +++ b/source/dnode/mgmt/mgmt_bnode/src/bmWorker.c @@ -47,10 +47,8 @@ static inline void bmSendRsp(SRpcMsg *pMsg, int32_t code) { static void bmProcessMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SBnodeMgmt *pMgmt = pInfo->ahandle; - + int32_t code = -1; dTrace("msg:%p, get from bnode-monitor queue", pMsg); - SRpcMsg *pRpc = pMsg; - int32_t code = -1; if (pMsg->msgType == TDMT_MON_BM_INFO) { code = bmProcessGetMonBmInfoReq(pMgmt, pMsg); @@ -58,13 +56,13 @@ static void bmProcessMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { terrno = TSDB_CODE_MSG_NOT_PROCESSED; } - if (pRpc->msgType & 1U) { + if (IsReq(pMsg)) { if (code != 0 && terrno != 0) code = terrno; bmSendRsp(pMsg, code); } dTrace("msg:%p, is freed, code:0x%x", pMsg, code); - rpcFreeCont(pRpc->pCont); + rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index 6a7e0ad322..bd78fe46ef 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -19,7 +19,6 @@ static void *dmStatusThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); - setThreadName("dnode-status"); while (1) { @@ -40,7 +39,6 @@ static void *dmStatusThreadFp(void *param) { static void *dmMonitorThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); - setThreadName("dnode-monitor"); while (1) { @@ -103,11 +101,9 @@ void dmStopMonitorThread(SDnodeMgmt *pMgmt) { static void dmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SDnodeMgmt *pMgmt = pInfo->ahandle; int32_t code = -1; - tmsg_t msgType = pMsg->msgType; - bool isRequest = msgType & 1u; - dTrace("msg:%p, will be processed in dnode-mgmt queue, type:%s", pMsg, TMSG_INFO(msgType)); + dTrace("msg:%p, will be processed in dnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); - switch (msgType) { + switch (pMsg->msgType) { case TDMT_DND_CONFIG_DNODE: code = dmProcessConfigReq(pMgmt, pMsg); break; @@ -149,7 +145,7 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { break; } - if (isRequest) { + if (IsReq(pMsg)) { if (code != 0 && terrno != 0) code = terrno; SRpcMsg rsp = { .code = code, diff --git a/source/dnode/mgmt/mgmt_snode/src/smWorker.c b/source/dnode/mgmt/mgmt_snode/src/smWorker.c index fcfc4f4cee..34a205232e 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smWorker.c +++ b/source/dnode/mgmt/mgmt_snode/src/smWorker.c @@ -28,10 +28,8 @@ static inline void smSendRsp(SRpcMsg *pMsg, int32_t code) { static void smProcessMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SSnodeMgmt *pMgmt = pInfo->ahandle; - + int32_t code = -1; dTrace("msg:%p, get from snode-monitor queue", pMsg); - SRpcMsg *pRpc = pMsg; - int32_t code = -1; if (pMsg->msgType == TDMT_MON_SM_INFO) { code = smProcessGetMonitorInfoReq(pMgmt, pMsg); @@ -39,13 +37,13 @@ static void smProcessMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { terrno = TSDB_CODE_MSG_NOT_PROCESSED; } - if (pRpc->msgType & 1U) { + if (IsReq(pMsg)) { if (code != 0 && terrno != 0) code = terrno; smSendRsp(pMsg, code); } dTrace("msg:%p, is freed, code:0x%x", pMsg, code); - rpcFreeCont(pRpc->pCont); + rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); } diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 3a5e8a671c..e7bfbdae58 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -29,7 +29,7 @@ static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) { tmsgSendRsp(&rsp); } -static void vmProcessMgmtMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { +static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SVnodeMgmt *pMgmt = pInfo->ahandle; int32_t code = -1; dTrace("msg:%p, get from vnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); @@ -92,7 +92,7 @@ static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; - SArray * pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *)); + SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *)); if (pArray == NULL) { dError("failed to process %d msgs in write-queue since %s", numOfMsgs, terrstr()); return; @@ -222,8 +222,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO } static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) { - SRpcMsg * pRpc = pMsg; - SMsgHead *pHead = pRpc->pCont; + SMsgHead *pHead = pMsg->pCont; int32_t code = 0; pHead->contLen = ntohl(pHead->contLen); @@ -237,23 +236,23 @@ static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType switch (qtype) { case QUERY_QUEUE: - dTrace("msg:%p, put into vnode-query worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, put into vnode-query worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pQueryQ, pMsg); break; case FETCH_QUEUE: - dTrace("msg:%p, put into vnode-fetch worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, put into vnode-fetch worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pFetchQ, pMsg); break; case WRITE_QUEUE: - dTrace("msg:%p, put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pWriteQ, pMsg); break; case SYNC_QUEUE: - dTrace("msg:%p, put into vnode-sync worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, put into vnode-sync worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pSyncQ, pMsg); break; case MERGE_QUEUE: - dTrace("msg:%p, put into vnode-merge worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); + dTrace("msg:%p, put into vnode-merge worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType)); taosWriteQitem(pVnode->pMergeQ, pMsg); break; default: @@ -301,7 +300,7 @@ int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { } static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) { - SMsgHead * pHead = pRpc->pCont; + SMsgHead *pHead = pRpc->pCont; SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); if (pVnode == NULL) return -1; @@ -469,7 +468,7 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { .min = 1, .max = 1, .name = "vnode-mgmt", - .fp = (FItem)vmProcessMgmtMonitorQueue, + .fp = (FItem)vmProcessQueue, .param = pMgmt, }; if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) { @@ -481,7 +480,7 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { .min = 1, .max = 1, .name = "vnode-monitor", - .fp = (FItem)vmProcessMgmtMonitorQueue, + .fp = (FItem)vmProcessQueue, .param = pMgmt, }; if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) { diff --git a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h index 2b3ac7ae73..27f1140f23 100644 --- a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h +++ b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h @@ -137,7 +137,6 @@ SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper); void dmSetStatus(SDnode *pDnode, EDndRunStatus stype); void dmProcessServerStartupStatus(SDnode *pDnode, SRpcMsg *pMsg); void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pMsg); -void dmProcessFetchRsp(SRpcMsg *pMsg); // dmNodes.c int32_t dmOpenNode(SMgmtWrapper *pWrapper); diff --git a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c index c83550c7b1..787f5e5019 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c +++ b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c @@ -314,8 +314,3 @@ void dmProcessServerStartupStatus(SDnode *pDnode, SRpcMsg *pMsg) { rpcSendResponse(&rsp); rpcFreeCont(pMsg->pCont); } - -void dmProcessFetchRsp(SRpcMsg *pMsg) { - qWorkerProcessFetchRsp(NULL, NULL, pMsg); - // rpcFreeCont(pMsg->pCont); -} \ No newline at end of file diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 8dfcb798ac..6fbfae8b41 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -15,6 +15,7 @@ #define _DEFAULT_SOURCE #include "dmMgmt.h" +#include "qworker.h" static void dmSendRedirectRsp(SRpcMsg *pMsg, const SEpSet *pNewEpSet); static void dmSendRsp(SRpcMsg *pMsg); @@ -61,7 +62,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { dmProcessNetTestReq(pDnode, pRpc); return; } else if (pRpc->msgType == TDMT_MND_SYSTABLE_RETRIEVE_RSP || pRpc->msgType == TDMT_VND_FETCH_RSP) { - dmProcessFetchRsp(pRpc); + qWorkerProcessFetchRsp(NULL, NULL, pRpc); return; } else { } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 297b518ac7..88cc97ddd5 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -147,9 +147,6 @@ _err: int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { vTrace("message in vnode query queue is processing"); -#if 0 - SReadHandle handle = {.reader = pVnode->pTsdb, .meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb}; -#endif SReadHandle handle = {.meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb}; switch (pMsg->msgType) { case TDMT_VND_QUERY: -- GitLab