diff --git a/source/dnode/mgmt/main/inc/dnd.h b/source/dnode/mgmt/main/inc/dnd.h index 255a739add..f3a409f257 100644 --- a/source/dnode/mgmt/main/inc/dnd.h +++ b/source/dnode/mgmt/main/inc/dnd.h @@ -134,33 +134,42 @@ typedef struct SDnode { SMgmtWrapper wrappers[NODE_MAX]; } SDnode; +// dndFile.h +int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed); +int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed); + +// dndInt.h +EDndStatus dndGetStatus(SDnode *pDnode); +void dndSetStatus(SDnode *pDnode, EDndStatus stat); +void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId); +SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType); +int32_t dndMarkWrapper(SMgmtWrapper *pWrapper); +void dndReleaseWrapper(SMgmtWrapper *pWrapper); + +// dndMonitor.h +void dndSendMonitorReport(SDnode *pDnode); + +// dndMsg.h +void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc); +int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg); + +// dndStr.h +const char *dndStatStr(EDndStatus stat); const char *dndNodeLogStr(ENodeType ntype); const char *dndNodeProcStr(ENodeType ntype); const char *dndEventStr(EDndEvent ev); -EDndStatus dndGetStatus(SDnode *pDnode); -void dndSetStatus(SDnode *pDnode, EDndStatus stat); -void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId); -void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc); -void dndSendMonitorReport(SDnode *pDnode); +// dndTransport.h int32_t dndInitServer(SDnode *pDnode); void dndCleanupServer(SDnode *pDnode); int32_t dndInitClient(SDnode *pDnode); void dndCleanupClient(SDnode *pDnode); -int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg); int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pMsg); void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp); void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper); -int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed); -int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed); - -SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType nodeType); -int32_t dndMarkWrapper(SMgmtWrapper *pWrapper); -void dndReleaseWrapper(SMgmtWrapper *pWrapper); - #ifdef __cplusplus } #endif diff --git a/source/dnode/mgmt/main/inc/dndInt.h b/source/dnode/mgmt/main/inc/dndInt.h index 612d35d513..0c4ada1df3 100644 --- a/source/dnode/mgmt/main/inc/dndInt.h +++ b/source/dnode/mgmt/main/inc/dndInt.h @@ -29,26 +29,24 @@ extern "C" { #endif -// dndInt.c -int32_t dndInit(); -void dndCleanup(); -const char *dndStatStr(EDndStatus stat); -void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup); -void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg); +// dndEnv.h +int32_t dndInit(); +void dndCleanup(); -// dndMsg.c -void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, SEpSet *pEpSet); - -// dndExec.c +// dndExec.h int32_t dndOpenNode(SMgmtWrapper *pWrapper); void dndCloseNode(SMgmtWrapper *pWrapper); int32_t dndRun(SDnode *pDnode); -// dndObj.c +// dndInt.c SDnode *dndCreate(const SDnodeOpt *pOption); void dndClose(SDnode *pDnode); void dndHandleEvent(SDnode *pDnode, EDndEvent event); +// dndMsg.c +void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, SEpSet *pEpSet); +void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg); + // dndTransport.c int32_t dndInitMsgHandle(SDnode *pDnode); void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp); diff --git a/source/dnode/mgmt/main/src/dndEnv.c b/source/dnode/mgmt/main/src/dndEnv.c index 8792147822..9f75594335 100644 --- a/source/dnode/mgmt/main/src/dndEnv.c +++ b/source/dnode/mgmt/main/src/dndEnv.c @@ -57,40 +57,3 @@ void dndCleanup() { taosStopCacheRefreshWorker(); dInfo("dnode env is cleaned up"); } - -void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId) { - pWrapper->msgFps[TMSG_INDEX(msgType)] = nodeMsgFp; - pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId; -} - -EDndStatus dndGetStatus(SDnode *pDnode) { return pDnode->status; } - -void dndSetStatus(SDnode *pDnode, EDndStatus status) { - if (pDnode->status != status) { - dDebug("dnode status set from %s to %s", dndStatStr(pDnode->status), dndStatStr(status)); - pDnode->status = status; - } -} - -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); -} diff --git a/source/dnode/mgmt/main/src/dndInt.c b/source/dnode/mgmt/main/src/dndInt.c index 6f6e21b983..25e4d89c5b 100644 --- a/source/dnode/mgmt/main/src/dndInt.c +++ b/source/dnode/mgmt/main/src/dndInt.c @@ -188,4 +188,18 @@ void dndReleaseWrapper(SMgmtWrapper *pWrapper) { int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1); taosRUnLockLatch(&pWrapper->latch); dTrace("node:%s, is released, refCount:%d", pWrapper->name, refCount); -} \ No newline at end of file +} + +void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId) { + pWrapper->msgFps[TMSG_INDEX(msgType)] = nodeMsgFp; + pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId; +} + +EDndStatus dndGetStatus(SDnode *pDnode) { return pDnode->status; } + +void dndSetStatus(SDnode *pDnode, EDndStatus status) { + if (pDnode->status != status) { + dDebug("dnode status set from %s to %s", dndStatStr(pDnode->status), dndStatStr(status)); + pDnode->status = status; + } +} diff --git a/source/dnode/mgmt/main/src/dndMsg.c b/source/dnode/mgmt/main/src/dndMsg.c index 2ab1e401c7..bb0f8ccb89 100644 --- a/source/dnode/mgmt/main/src/dndMsg.c +++ b/source/dnode/mgmt/main/src/dndMsg.c @@ -66,8 +66,8 @@ void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSet) { dTrace("msg:%p, is created, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, pMsg->user); code = (*msgFp)(pWrapper, pMsg); } else if (pWrapper->procType == PROC_PARENT) { - dTrace("msg:%p, is created and put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle, - pRpc->ahandle, pMsg->user); + dTrace("msg:%p, is created and put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, + pMsg->user); code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, PROC_REQ); } else { dTrace("msg:%p, should not processed in child process, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, @@ -171,4 +171,27 @@ int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg) { terrno = TSDB_CODE_MSG_NOT_PROCESSED; return -1; } -} \ No newline at end of file +} + +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); +}