diff --git a/source/dnode/mgmt/dm/src/dmMsg.c b/source/dnode/mgmt/dm/src/dmMsg.c index 90117044f9..979db7aac0 100644 --- a/source/dnode/mgmt/dm/src/dmMsg.c +++ b/source/dnode/mgmt/dm/src/dmMsg.c @@ -117,6 +117,107 @@ int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { return TSDB_CODE_OPS_NOT_SUPPORT; } + +static int32_t dndProcessCreateNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) { + SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); + if (pWrapper != NULL) { + dndReleaseWrapper(pWrapper); + terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED; + dError("failed to create node since %s", terrstr()); + return -1; + } + + pWrapper = &pDnode->wrappers[ntype]; + + if (taosMkDir(pWrapper->path) != 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + dError("failed to create dir:%s since %s", pWrapper->path, terrstr()); + return -1; + } + + int32_t code = (*pWrapper->fp.createMsgFp)(pWrapper, pMsg); + if (code != 0) { + dError("node:%s, failed to open since %s", pWrapper->name, terrstr()); + } else { + dDebug("node:%s, has been opened", pWrapper->name); + pWrapper->deployed = true; + } + + return code; +} + +static int32_t dndProcessDropNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) { + SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); + if (pWrapper == NULL) { + terrno = TSDB_CODE_NODE_NOT_DEPLOYED; + dError("failed to drop node since %s", terrstr()); + return -1; + } + + taosWLockLatch(&pWrapper->latch); + pWrapper->deployed = false; + + int32_t code = (*pWrapper->fp.dropMsgFp)(pWrapper, pMsg); + if (code != 0) { + pWrapper->deployed = true; + dError("node:%s, failed to drop since %s", pWrapper->name, terrstr()); + } else { + pWrapper->deployed = false; + dDebug("node:%s, has been dropped", pWrapper->name); + } + + taosWUnLockLatch(&pWrapper->latch); + dndReleaseWrapper(pWrapper); + return code; +} + +int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg) { + switch (pMsg->rpcMsg.msgType) { + case TDMT_DND_CREATE_MNODE: + return dndProcessCreateNodeMsg(pDnode, MNODE, pMsg); + case TDMT_DND_DROP_MNODE: + return dndProcessDropNodeMsg(pDnode, MNODE, pMsg); + case TDMT_DND_CREATE_QNODE: + return dndProcessCreateNodeMsg(pDnode, QNODE, pMsg); + case TDMT_DND_DROP_QNODE: + return dndProcessDropNodeMsg(pDnode, QNODE, pMsg); + case TDMT_DND_CREATE_SNODE: + return dndProcessCreateNodeMsg(pDnode, SNODE, pMsg); + case TDMT_DND_DROP_SNODE: + return dndProcessDropNodeMsg(pDnode, SNODE, pMsg); + case TDMT_DND_CREATE_BNODE: + return dndProcessCreateNodeMsg(pDnode, BNODE, pMsg); + case TDMT_DND_DROP_BNODE: + return dndProcessDropNodeMsg(pDnode, BNODE, pMsg); + default: + terrno = TSDB_CODE_MSG_NOT_PROCESSED; + return -1; + } +} + +void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) { + SStartupReq *pStartup = &pDnode->startup; + tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN); + tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN); + pStartup->finished = 0; +} + +void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) { + memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq)); + pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING); +} + +void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) { + dDebug("startup req is received"); + SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq)); + dndGetStartup(pDnode, pStartup); + + dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished); + SRpcMsg rpcRsp = { + .handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle}; + rpcSendResponse(&rpcRsp); +} + void dmInitMsgHandles(SMgmtWrapper *pWrapper) { // Requests handled by DNODE dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmProcessMgmtMsg, VND_VGID); diff --git a/source/dnode/mgmt/main/src/dndMsg.c b/source/dnode/mgmt/main/src/dndMsg.c deleted file mode 100644 index 5d739b7e15..0000000000 --- a/source/dnode/mgmt/main/src/dndMsg.c +++ /dev/null @@ -1,117 +0,0 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * This program is free software: you can use, redistribute, and/or modify - * it under the terms of the GNU Affero General Public License, version 3 - * or later ("AGPL"), as published by the Free Software Foundation. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ - -#define _DEFAULT_SOURCE -#include "dndInt.h" - -static int32_t dndProcessCreateNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) { - SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); - if (pWrapper != NULL) { - dndReleaseWrapper(pWrapper); - terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED; - dError("failed to create node since %s", terrstr()); - return -1; - } - - pWrapper = &pDnode->wrappers[ntype]; - - if (taosMkDir(pWrapper->path) != 0) { - terrno = TAOS_SYSTEM_ERROR(errno); - dError("failed to create dir:%s since %s", pWrapper->path, terrstr()); - return -1; - } - - int32_t code = (*pWrapper->fp.createMsgFp)(pWrapper, pMsg); - if (code != 0) { - dError("node:%s, failed to open since %s", pWrapper->name, terrstr()); - } else { - dDebug("node:%s, has been opened", pWrapper->name); - pWrapper->deployed = true; - } - - return code; -} - -static int32_t dndProcessDropNodeMsg(SDnode *pDnode, ENodeType ntype, SNodeMsg *pMsg) { - SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); - if (pWrapper == NULL) { - terrno = TSDB_CODE_NODE_NOT_DEPLOYED; - dError("failed to drop node since %s", terrstr()); - return -1; - } - - taosWLockLatch(&pWrapper->latch); - pWrapper->deployed = false; - - int32_t code = (*pWrapper->fp.dropMsgFp)(pWrapper, pMsg); - if (code != 0) { - pWrapper->deployed = true; - dError("node:%s, failed to drop since %s", pWrapper->name, terrstr()); - } else { - pWrapper->deployed = false; - dDebug("node:%s, has been dropped", pWrapper->name); - } - - taosWUnLockLatch(&pWrapper->latch); - dndReleaseWrapper(pWrapper); - return code; -} - -int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg) { - switch (pMsg->rpcMsg.msgType) { - case TDMT_DND_CREATE_MNODE: - return dndProcessCreateNodeMsg(pDnode, MNODE, pMsg); - case TDMT_DND_DROP_MNODE: - return dndProcessDropNodeMsg(pDnode, MNODE, pMsg); - case TDMT_DND_CREATE_QNODE: - return dndProcessCreateNodeMsg(pDnode, QNODE, pMsg); - case TDMT_DND_DROP_QNODE: - return dndProcessDropNodeMsg(pDnode, QNODE, pMsg); - case TDMT_DND_CREATE_SNODE: - return dndProcessCreateNodeMsg(pDnode, SNODE, pMsg); - case TDMT_DND_DROP_SNODE: - return dndProcessDropNodeMsg(pDnode, SNODE, pMsg); - case TDMT_DND_CREATE_BNODE: - return dndProcessCreateNodeMsg(pDnode, BNODE, pMsg); - case TDMT_DND_DROP_BNODE: - return dndProcessDropNodeMsg(pDnode, BNODE, pMsg); - default: - terrno = TSDB_CODE_MSG_NOT_PROCESSED; - return -1; - } -} - -void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) { - SStartupReq *pStartup = &pDnode->startup; - tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN); - tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN); - pStartup->finished = 0; -} - -void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) { - memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq)); - pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING); -} - -void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) { - dDebug("startup req is received"); - SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq)); - dndGetStartup(pDnode, pStartup); - - dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished); - SRpcMsg rpcRsp = { - .handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle}; - rpcSendResponse(&rpcRsp); -}