From 08b087ec9132f57c022d3aeb7e21529fb16ba657 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sat, 30 May 2020 10:47:57 +0000 Subject: [PATCH] add a queue for mgmt message from mnode --- src/dnode/inc/dnodeMgmt.h | 2 +- src/dnode/src/dnodeMgmt.c | 104 +++++++++++++++++++++++++++++++------- src/dnode/src/dnodePeer.c | 8 +-- 3 files changed, 91 insertions(+), 23 deletions(-) diff --git a/src/dnode/inc/dnodeMgmt.h b/src/dnode/inc/dnodeMgmt.h index 4d15dc5a86..826f4ff1c1 100644 --- a/src/dnode/inc/dnodeMgmt.h +++ b/src/dnode/inc/dnodeMgmt.h @@ -22,7 +22,7 @@ extern "C" { int32_t dnodeInitMgmt(); void dnodeCleanupMgmt(); -void dnodeDispatchToDnodeMgmt(SRpcMsg *rpcMsg); +void dnodeDispatchToMgmtQueue(SRpcMsg *rpcMsg); void* dnodeGetVnode(int32_t vgId); int32_t dnodeGetVnodeStatus(void *pVnode); diff --git a/src/dnode/src/dnodeMgmt.c b/src/dnode/src/dnodeMgmt.c index 14c2a725d9..1dd6952c6e 100644 --- a/src/dnode/src/dnodeMgmt.c +++ b/src/dnode/src/dnodeMgmt.c @@ -22,6 +22,7 @@ #include "ttimer.h" #include "tsdb.h" #include "twal.h" +#include "tqueue.h" #include "tsync.h" #include "ttime.h" #include "ttimer.h" @@ -46,6 +47,9 @@ static SRpcIpSet tsDMnodeIpSetForPeer = {0}; static SRpcIpSet tsDMnodeIpSetForShell = {0}; static SDMMnodeInfos tsDMnodeInfos = {0}; static SDMDnodeCfg tsDnodeCfg = {0}; +static taos_qset tsMgmtQset = NULL; +static taos_queue tsMgmtQueue = NULL; +static pthread_t tsQthread; static void dnodeUpdateMnodeInfos(SDMMnodeInfos *pMnodes); static bool dnodeReadMnodeInfos(); @@ -55,6 +59,7 @@ static bool dnodeReadDnodeCfg(); static void dnodeSaveDnodeCfg(); static void dnodeProcessStatusRsp(SRpcMsg *pMsg); static void dnodeSendStatusMsg(void *handle, void *tmrId); +static void *dnodeProcessMgmtQueue(void *param); static int32_t dnodeOpenVnodes(); static void dnodeCloseVnodes(); @@ -74,12 +79,6 @@ int32_t dnodeInitMgmt() { dnodeReadDnodeCfg(); tsRebootTime = taosGetTimestampSec(); - tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM"); - if (tsDnodeTmr == NULL) { - dError("failed to init dnode timer"); - return -1; - } - if (!dnodeReadMnodeInfos()) { memset(&tsDMnodeIpSetForPeer, 0, sizeof(SRpcIpSet)); memset(&tsDMnodeIpSetForShell, 0, sizeof(SRpcIpSet)); @@ -118,8 +117,45 @@ int32_t dnodeInitMgmt() { } } - int32_t code = dnodeOpenVnodes(); + // create the queue and thread to handle the message + tsMgmtQset = taosOpenQset(); + if (tsMgmtQset == NULL) { + dError("failed to create the mgmt queue set"); + dnodeCleanupMgmt(); + return -1; + } + + tsMgmtQueue = taosOpenQueue(); + if (tsMgmtQueue == NULL) { + dError("failed to create the mgmt queue"); + dnodeCleanupMgmt(); + return -1; + } + + taosAddIntoQset(tsMgmtQset, tsMgmtQueue, NULL); + + pthread_attr_t thAttr; + pthread_attr_init(&thAttr); + pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); + + int32_t code = pthread_create(&tsQthread, &thAttr, dnodeProcessMgmtQueue, NULL); + pthread_attr_destroy(&thAttr); + if (code != 0) { + dError("failed to create thread to process mgmt queue, reason:%s", strerror(errno)); + dnodeCleanupMgmt(); + return -1; + } + + code = dnodeOpenVnodes(); if (code != TSDB_CODE_SUCCESS) { + dnodeCleanupMgmt(); + return -1; + } + + tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM"); + if (tsDnodeTmr == NULL) { + dError("failed to init dnode timer"); + dnodeCleanupMgmt(); return -1; } @@ -142,22 +178,54 @@ void dnodeCleanupMgmt() { } dnodeCloseVnodes(); + + if (tsMgmtQset) taosQsetThreadResume(tsMgmtQset); + if (tsQthread) pthread_join(tsQthread, NULL); + + if (tsMgmtQueue) taosCloseQueue(tsMgmtQueue); + if (tsMgmtQset) taosCloseQset(tsMgmtQset); + tsMgmtQset = NULL; + tsMgmtQueue = NULL; + } -void dnodeDispatchToDnodeMgmt(SRpcMsg *pMsg) { - SRpcMsg rsp; +void dnodeDispatchToMgmtQueue(SRpcMsg *pMsg) { + void *item; - if (dnodeProcessMgmtMsgFp[pMsg->msgType]) { - rsp.code = (*dnodeProcessMgmtMsgFp[pMsg->msgType])(pMsg); - } else { - rsp.code = TSDB_CODE_MSG_NOT_PROCESSED; - } + item = taosAllocateQitem(sizeof(SRpcMsg)); + memcpy(item, pMsg, sizeof(SRpcMsg)); + + taosWriteQitem(tsMgmtQueue, 1, item); +} + +static void *dnodeProcessMgmtQueue(void *param) { + SRpcMsg *pMsg; + SRpcMsg rsp; + int type; + void *handle; - rsp.handle = pMsg->handle; - rsp.pCont = NULL; - rpcSendResponse(&rsp); + while (1) { + if (taosReadQitemFromQset(tsMgmtQset, &type, (void **) &pMsg, &handle) == 0) { + dTrace("dnode mgmt got no message from qset, exit ..."); + break; + } + + dTrace("%p, msg:%s will be processed", pMsg->ahandle, taosMsg[pMsg->msgType]); + if (dnodeProcessMgmtMsgFp[pMsg->msgType]) { + rsp.code = (*dnodeProcessMgmtMsgFp[pMsg->msgType])(pMsg); + } else { + rsp.code = TSDB_CODE_MSG_NOT_PROCESSED; + } + + rsp.handle = pMsg->handle; + rsp.pCont = NULL; + rpcSendResponse(&rsp); + + rpcFreeCont(pMsg->pCont); + taosFreeQitem(pMsg); + } - rpcFreeCont(pMsg->pCont); + return NULL; } static int32_t dnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { diff --git a/src/dnode/src/dnodePeer.c b/src/dnode/src/dnodePeer.c index 9a7b0837e8..ea3af08d71 100644 --- a/src/dnode/src/dnodePeer.c +++ b/src/dnode/src/dnodePeer.c @@ -43,10 +43,10 @@ int32_t dnodeInitServer() { dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE] = dnodeDispatchToVnodeWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE] = dnodeDispatchToVnodeWriteQueue; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToDnodeMgmt; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeDispatchToDnodeMgmt; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToDnodeMgmt; - dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToDnodeMgmt; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToMgmtQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeDispatchToMgmtQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToMgmtQueue; + dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMnodePeerQueue; -- GitLab