提交 f4c31418 编写于 作者: S Shengliang Guan

refactor: rpc msg handler

上级 12d27bfd
......@@ -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);
}
......
......@@ -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,
......
......@@ -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);
}
......
......@@ -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) {
......
......@@ -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);
......
......@@ -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
......@@ -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 {
}
......
......@@ -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:
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册