From a27bffd6b20699a4475cda41424de1f22ea9fae6 Mon Sep 17 00:00:00 2001 From: Shengliang Date: Wed, 11 May 2022 17:46:21 +0800 Subject: [PATCH] refactor: node mgmt --- source/dnode/mgmt/mgmt_qnode/inc/qmInt.h | 25 ++++----- source/dnode/mgmt/mgmt_qnode/src/qmHandle.c | 57 ++++++++++++-------- source/dnode/mgmt/mgmt_qnode/src/qmInt.c | 59 ++++++++++----------- source/dnode/mgmt/mgmt_qnode/src/qmWorker.c | 36 +++++-------- 4 files changed, 88 insertions(+), 89 deletions(-) diff --git a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h index d52fbff683..46b7552e38 100644 --- a/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h +++ b/source/dnode/mgmt/mgmt_qnode/inc/qmInt.h @@ -26,30 +26,31 @@ extern "C" { typedef struct SQnodeMgmt { SQnode *pQnode; - SDnode *pDnode; - SMgmtWrapper *pWrapper; + SMsgCb msgCb; const char *path; + const char *name; + int32_t dnodeId; SSingleWorker queryWorker; SSingleWorker fetchWorker; SSingleWorker monitorWorker; } SQnodeMgmt; // qmHandle.c -void qmInitMsgHandle(SMgmtWrapper *pWrapper); -int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); -int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); -int32_t qmProcessGetMonQmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq); +SArray *qmGetMsgHandles(); +int32_t qmProcessCreateReq(SQnodeMgmt *pMgmt, SNodeMsg *pMsg); +int32_t qmProcessDropReq(SQnodeMgmt *pMgmt, SNodeMsg *pMsg); +int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SNodeMsg *pReq); // qmWorker.c -int32_t qmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); -int32_t qmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); -int32_t qmGetQueueSize(SMgmtWrapper *pWrapper, int32_t vgId, EQueueType qtype); +int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg); +int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype); int32_t qmStartWorker(SQnodeMgmt *pMgmt); void qmStopWorker(SQnodeMgmt *pMgmt); -int32_t qmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); -int32_t qmProcessFetchMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); -int32_t qmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); +int32_t qmPutNodeMsgToQueryQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg); +int32_t qmPutNodeMsgToFetchQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg); +int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg); #ifdef __cplusplus } diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmHandle.c b/source/dnode/mgmt/mgmt_qnode/src/qmHandle.c index e641dbf6ed..e7290cb369 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmHandle.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmHandle.c @@ -16,12 +16,12 @@ #define _DEFAULT_SOURCE #include "qmInt.h" -void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo) {} +static void qmGetMonitorInfo(SQnodeMgmt *pMgmt, SMonQmInfo *qmInfo) {} -int32_t qmProcessGetMonQmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) { +int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SNodeMsg *pReq) { SMonQmInfo qmInfo = {0}; - qmGetMonitorInfo(pWrapper, &qmInfo); - dmGetMonitorSysInfo(&qmInfo.sys); + qmGetMonitorInfo(pMgmt, &qmInfo); + dmGetSystemInfo(&qmInfo.sys); monGetLogs(&qmInfo.log); int32_t rspLen = tSerializeSMonQmInfo(NULL, 0, &qmInfo); @@ -43,8 +43,7 @@ int32_t qmProcessGetMonQmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) { return 0; } -int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SDnode *pDnode = pWrapper->pDnode; +int32_t qmProcessCreateReq(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { SRpcMsg *pReq = &pMsg->rpcMsg; SDCreateQnodeReq createReq = {0}; @@ -53,14 +52,14 @@ int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { return -1; } - if (createReq.dnodeId != pDnode->data.dnodeId) { + if (createReq.dnodeId != pMgmt->dnodeId) { terrno = TSDB_CODE_INVALID_OPTION; dError("failed to create qnode since %s", terrstr()); return -1; } bool deployed = true; - if (dmWriteFile(pWrapper, deployed) != 0) { + if (dmWriteFile(pMgmt->path, pMgmt->name, deployed) != 0) { dError("failed to write qnode file since %s", terrstr()); return -1; } @@ -68,8 +67,7 @@ int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { return 0; } -int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SDnode *pDnode = pWrapper->pDnode; +int32_t qmProcessDropReq(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { SRpcMsg *pReq = &pMsg->rpcMsg; SDDropQnodeReq dropReq = {0}; @@ -78,14 +76,14 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { return -1; } - if (dropReq.dnodeId != pDnode->data.dnodeId) { + if (dropReq.dnodeId != pMgmt->dnodeId) { terrno = TSDB_CODE_INVALID_OPTION; dError("failed to drop qnode since %s", terrstr()); return -1; } bool deployed = false; - if (dmWriteFile(pWrapper, deployed) != 0) { + if (dmWriteFile(pMgmt->path, pMgmt->name, deployed) != 0) { dError("failed to write qnode file since %s", terrstr()); return -1; } @@ -93,18 +91,31 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { return 0; } -void qmInitMsgHandle(SMgmtWrapper *pWrapper) { - dmSetMsgHandle(pWrapper, TDMT_MON_QM_INFO, qmProcessMonitorMsg, 0); +SArray *qmGetMsgHandles() { + int32_t code = -1; + SArray *pArray = taosArrayInit(16, sizeof(SMgmtHandle)); + if (pArray == NULL) goto _OVER; + + if (dmSetMgmtHandle(pArray, TDMT_MON_QM_INFO, qmPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER; // Requests handled by VNODE - dmSetMsgHandle(pWrapper, TDMT_VND_QUERY, qmProcessQueryMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, qmProcessQueryMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_FETCH, qmProcessFetchMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, qmProcessFetchMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, qmProcessFetchMsg, 1); + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, qmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, qmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH_RSP, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; - dmSetMsgHandle(pWrapper, TDMT_VND_RES_READY, qmProcessFetchMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_TASKS_STATUS, qmProcessFetchMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_CANCEL_TASK, qmProcessFetchMsg, 1); - dmSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, qmProcessFetchMsg, 1); + if (dmSetMgmtHandle(pArray, TDMT_VND_RES_READY, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_TASKS_STATUS, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, qmPutNodeMsgToFetchQueue, 1) == NULL) goto _OVER; + + code = 0; +_OVER: + if (code != 0) { + taosArrayDestroy(pArray); + return NULL; + } else { + return pArray; + } } diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c index 6837acccd2..6fef75f4b5 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmInt.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmInt.c @@ -16,21 +16,13 @@ #define _DEFAULT_SOURCE #include "qmInt.h" -static int32_t qmRequire(SMgmtWrapper *pWrapper, bool *required) { return dmReadFile(pWrapper, required); } - -static void qmInitOption(SQnodeMgmt *pMgmt, SQnodeOpt *pOption) { - SMsgCb msgCb = pMgmt->pDnode->data.msgCb; - msgCb.pWrapper = pMgmt->pWrapper; - msgCb.queueFps[QUERY_QUEUE] = qmPutMsgToQueryQueue; - msgCb.queueFps[FETCH_QUEUE] = qmPutMsgToFetchQueue; - msgCb.qsizeFp = qmGetQueueSize; - pOption->msgCb = msgCb; +static int32_t qmRequire(const SMgmtInputOpt *pInput, bool *required) { + return dmReadFile(pInput->path, pInput->name, required); } -static void qmClose(SMgmtWrapper *pWrapper) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; - if (pMgmt == NULL) return; +static void qmInitOption(SQnodeMgmt *pMgmt, SQnodeOpt *pOption) { pOption->msgCb = pMgmt->msgCb; } +static void qmClose(SQnodeMgmt *pMgmt) { dInfo("qnode-mgmt start to cleanup"); if (pMgmt->pQnode != NULL) { qmStopWorker(pMgmt); @@ -38,12 +30,11 @@ static void qmClose(SMgmtWrapper *pWrapper) { pMgmt->pQnode = NULL; } - pWrapper->pMgmt = NULL; taosMemoryFree(pMgmt); dInfo("qnode-mgmt is cleaned up"); } -static int32_t qmOpen(SMgmtWrapper *pWrapper) { +static int32_t qmOpen(const SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { dInfo("qnode-mgmt start to init"); SQnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SQnodeMgmt)); if (pMgmt == NULL) { @@ -51,41 +42,45 @@ static int32_t qmOpen(SMgmtWrapper *pWrapper) { return -1; } - pMgmt->path = pWrapper->path; - pMgmt->pDnode = pWrapper->pDnode; - pMgmt->pWrapper = pWrapper; - pWrapper->pMgmt = pMgmt; + pMgmt->path = pInput->path; + pMgmt->name = pInput->name; + pMgmt->dnodeId = pInput->dnodeId; + pMgmt->msgCb = pInput->msgCb; + pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qmPutRpcMsgToQueryQueue; + pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)qmPutRpcMsgToFetchQueue; + pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)qmGetQueueSize; + pMgmt->msgCb.pMgmt = pMgmt; SQnodeOpt option = {0}; qmInitOption(pMgmt, &option); pMgmt->pQnode = qndOpen(&option); if (pMgmt->pQnode == NULL) { dError("failed to open qnode since %s", terrstr()); - qmClose(pWrapper); + qmClose(pMgmt); return -1; } - dmReportStartup(pWrapper->pDnode, "qnode-impl", "initialized"); + tmsgReportStartup("qnode-impl", "initialized"); if (qmStartWorker(pMgmt) != 0) { dError("failed to start qnode worker since %s", terrstr()); - qmClose(pWrapper); + qmClose(pMgmt); return -1; } - dmReportStartup(pWrapper->pDnode, "qnode-worker", "initialized"); + tmsgReportStartup("qnode-worker", "initialized"); + pOutput->pMgmt = pMgmt; dInfo("qnode-mgmt is initialized"); return 0; } -void qmInitWrapper(SMgmtWrapper *pWrapper) { - SMgmtFp mgmtFp = {0}; - mgmtFp.openFp = qmOpen; - mgmtFp.closeFp = qmClose; - mgmtFp.createFp = qmProcessCreateReq; - mgmtFp.dropFp = qmProcessDropReq; - mgmtFp.requiredFp = qmRequire; +SMgmtFunc qmGetMgmtFunc() { + SMgmtFunc mgmtFunc = {0}; + mgmtFunc.openFp = qmOpen; + mgmtFunc.closeFp = (NodeCloseFp)qmClose; + mgmtFunc.createFp = (NodeCreateFp)qmProcessCreateReq; + mgmtFunc.dropFp = (NodeDropFp)qmProcessDropReq; + mgmtFunc.requiredFp = qmRequire; + mgmtFunc.getHandlesFp = qmGetMsgHandles; - qmInitMsgHandle(pWrapper); - pWrapper->name = "qnode"; - pWrapper->fp = mgmtFp; + return mgmtFunc; } diff --git a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c index 965d35cb3e..33df1b6786 100644 --- a/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c +++ b/source/dnode/mgmt/mgmt_qnode/src/qmWorker.c @@ -36,7 +36,7 @@ static void qmProcessMonitorQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { int32_t code = -1; if (pMsg->rpcMsg.msgType == TDMT_MON_QM_INFO) { - code = qmProcessGetMonQmInfoReq(pMgmt->pWrapper, pMsg); + code = qmProcessGetMonitorInfoReq(pMgmt, pMsg); } else { terrno = TSDB_CODE_MSG_NOT_PROCESSED; } @@ -83,27 +83,22 @@ static void qmProcessFetchQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { taosFreeQitem(pMsg); } -static void qmPutMsgToWorker(SSingleWorker *pWorker, SNodeMsg *pMsg) { +static int32_t qmPutNodeMsgToWorker(SSingleWorker *pWorker, SNodeMsg *pMsg) { dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); taosWriteQitem(pWorker->queue, pMsg); -} - -int32_t qmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; - qmPutMsgToWorker(&pMgmt->queryWorker, pMsg); return 0; } -int32_t qmProcessFetchMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; - qmPutMsgToWorker(&pMgmt->fetchWorker, pMsg); - return 0; +int32_t qmPutNodeMsgToQueryQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { + return qmPutNodeMsgToWorker(&pMgmt->queryWorker, pMsg); } -int32_t qmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; - qmPutMsgToWorker(&pMgmt->monitorWorker, pMsg); - return 0; +int32_t qmPutNodeMsgToFetchQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { + return qmPutNodeMsgToWorker(&pMgmt->fetchWorker, pMsg); +} + +int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SNodeMsg *pMsg) { + return qmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); } static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) { @@ -118,19 +113,16 @@ static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SR return 0; } -int32_t qmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; +int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { return qmPutRpcMsgToWorker(pMgmt, &pMgmt->queryWorker, pRpc); } -int32_t qmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) { - SQnodeMgmt *pMgmt = pWrapper->pMgmt; +int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) { return qmPutRpcMsgToWorker(pMgmt, &pMgmt->fetchWorker, pRpc); } -int32_t qmGetQueueSize(SMgmtWrapper *pWrapper, int32_t vgId, EQueueType qtype) { - int32_t size = -1; - SQnodeMgmt *pMgmt = pWrapper->pMgmt; +int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { + int32_t size = -1; switch (qtype) { case QUERY_QUEUE: