From 0094f30140f20dca2f7a6a062743918068ea2e7b Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 12 Apr 2022 15:27:17 +0800 Subject: [PATCH] refact(cluster): modify node management code --- source/dnode/mgmt/bm/bmHandle.c | 3 +- source/dnode/mgmt/inc/bmInt.h | 2 +- source/dnode/mgmt/inc/dndInt.h | 1 + source/dnode/mgmt/main/dndExec.c | 177 +++++++++++++++++++------------ source/dnode/mgmt/mm/mmHandle.c | 1 + source/dnode/mgmt/qm/qmHandle.c | 3 +- source/dnode/mgmt/sm/smHandle.c | 3 +- 7 files changed, 117 insertions(+), 73 deletions(-) diff --git a/source/dnode/mgmt/bm/bmHandle.c b/source/dnode/mgmt/bm/bmHandle.c index 645a1d09c2..d110592603 100644 --- a/source/dnode/mgmt/bm/bmHandle.c +++ b/source/dnode/mgmt/bm/bmHandle.c @@ -58,7 +58,7 @@ int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to create bnode since %s, input:%d cur:%d", terrstr(), createReq.dnodeId, pDnode->dnodeId); return -1; } else { - return bmOpen(pWrapper); + return dndOpenNode(pWrapper); } } @@ -77,6 +77,7 @@ int32_t bmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to drop bnode since %s", terrstr()); return -1; } else { + // dndCloseNode(pWrapper); return bmDrop(pWrapper); } } diff --git a/source/dnode/mgmt/inc/bmInt.h b/source/dnode/mgmt/inc/bmInt.h index 84a6a53e99..3158fe7d34 100644 --- a/source/dnode/mgmt/inc/bmInt.h +++ b/source/dnode/mgmt/inc/bmInt.h @@ -27,7 +27,7 @@ extern "C" { typedef struct SBnodeMgmt { SBnode *pBnode; SDnode *pDnode; - SMgmtWrapper *pWrapper; + SMgmtWrapper *pWrapper; const char *path; SMultiWorker writeWorker; SSingleWorker monitorWorker; diff --git a/source/dnode/mgmt/inc/dndInt.h b/source/dnode/mgmt/inc/dndInt.h index 7faf1e4276..a38fe87b59 100644 --- a/source/dnode/mgmt/inc/dndInt.h +++ b/source/dnode/mgmt/inc/dndInt.h @@ -126,6 +126,7 @@ typedef struct SDnode { int32_t numOfDisks; uint16_t serverPort; bool dropped; + EProcType procType; EDndType ntype; EDndStatus status; EDndEvent event; diff --git a/source/dnode/mgmt/main/dndExec.c b/source/dnode/mgmt/main/dndExec.c index 6c0d0456c9..51569d2ed4 100644 --- a/source/dnode/mgmt/main/dndExec.c +++ b/source/dnode/mgmt/main/dndExec.c @@ -27,7 +27,81 @@ static bool dndRequireNode(SMgmtWrapper *pWrapper) { return required; } -int32_t dndOpenNode(SMgmtWrapper *pWrapper) { +static int32_t dndInitNodeProc(SMgmtWrapper *pWrapper) { + int32_t shmsize = tsMnodeShmSize; + if (pWrapper->ntype == VNODES) { + shmsize = tsVnodeShmSize; + } else if (pWrapper->ntype == QNODE) { + shmsize = tsQnodeShmSize; + } else if (pWrapper->ntype == SNODE) { + shmsize = tsSnodeShmSize; + } else if (pWrapper->ntype == MNODE) { + shmsize = tsMnodeShmSize; + } else if (pWrapper->ntype == BNODE) { + shmsize = tsBnodeShmSize; + } else { + return -1; + } + + if (taosCreateShm(&pWrapper->shm, pWrapper->ntype, shmsize) != 0) { + terrno = TAOS_SYSTEM_ERROR(terrno); + dError("node:%s, failed to create shm size:%d since %s", pWrapper->name, shmsize, terrstr()); + return -1; + } + dInfo("node:%s, shm:%d is created, size:%d", pWrapper->name, pWrapper->shm.id, shmsize); + + SProcCfg cfg = dndGenProcCfg(pWrapper); + cfg.isChild = false; + pWrapper->procType = PROC_PARENT; + pWrapper->pProc = taosProcInit(&cfg); + if (pWrapper->pProc == NULL) { + dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr()); + return -1; + } + + return 0; +} + +static int32_t dndNewNodeProc(SMgmtWrapper *pWrapper, EDndType n) { + char tstr[8] = {0}; + char *args[6] = {0}; + snprintf(tstr, sizeof(tstr), "%d", n); + args[1] = "-c"; + args[2] = configDir; + args[3] = "-n"; + args[4] = tstr; + args[5] = NULL; + + int32_t pid = taosNewProc(args); + if (pid <= 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + dError("node:%s, failed to exec in new process since %s", pWrapper->name, terrstr()); + return -1; + } + + pWrapper->procId = pid; + dInfo("node:%s, continue running in new process:%d", pWrapper->name, pid); + return 0; +} + +static int32_t dndRunNodeProc(SMgmtWrapper *pWrapper) { + if (pWrapper->pDnode->ntype == NODE_MAX) { + dInfo("node:%s, should be started manually", pWrapper->name); + } else { + if (dndNewNodeProc(pWrapper, pWrapper->ntype) != 0) { + return -1; + } + } + + if (taosProcRun(pWrapper->pProc) != 0) { + dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr()); + return -1; + } + + return 0; +} + +static int32_t dndOpenNodeImp(SMgmtWrapper *pWrapper) { if (taosMkDir(pWrapper->path) != 0) { terrno = TAOS_SYSTEM_ERROR(errno); dError("node:%s, failed to create dir:%s since %s", pWrapper->name, pWrapper->path, terrstr()); @@ -44,7 +118,19 @@ int32_t dndOpenNode(SMgmtWrapper *pWrapper) { return 0; } -void dndCloseNode(SMgmtWrapper *pWrapper) { +int32_t dndOpenNode(SMgmtWrapper *pWrapper) { + SDnode *pDnode = pWrapper->pDnode; + if (pDnode->procType == PROC_SINGLE) { + return dndOpenNodeImp(pWrapper); + } else if (pDnode->procType == PROC_PARENT) { + if (dndInitNodeProc(pWrapper) != 0) return -1; + if (dndWriteShmFile(pDnode) != 0) return -1; + if (dndRunNodeProc(pWrapper) != 0) return -1; + } + return 0; +} + +static void dndCloseNodeImp(SMgmtWrapper *pWrapper) { dDebug("node:%s, mgmt start to close", pWrapper->name); pWrapper->required = false; taosWLockLatch(&pWrapper->latch); @@ -65,27 +151,17 @@ void dndCloseNode(SMgmtWrapper *pWrapper) { dDebug("node:%s, mgmt has been closed", pWrapper->name); } - -static int32_t dndNewProc(SMgmtWrapper *pWrapper, EDndType n) { - char tstr[8] = {0}; - char *args[6] = {0}; - snprintf(tstr, sizeof(tstr), "%d", n); - args[1] = "-c"; - args[2] = configDir; - args[3] = "-n"; - args[4] = tstr; - args[5] = NULL; - - int32_t pid = taosNewProc(args); - if (pid <= 0) { - terrno = TAOS_SYSTEM_ERROR(errno); - dError("node:%s, failed to exec in new process since %s", pWrapper->name, terrstr()); - return -1; +void dndCloseNode(SMgmtWrapper *pWrapper) { + if (pWrapper->pDnode->procType == PROC_PARENT) { + if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) { + dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId); + taosKillProc(pWrapper->procId); + dInfo("node:%s, wait for child process:%d to stop", pWrapper->name, pWrapper->procId); + taosWaitProc(pWrapper->procId); + dInfo("node:%s, child process:%d is stopped", pWrapper->name, pWrapper->procId); + } } - - pWrapper->procId = pid; - dInfo("node:%s, continue running in new process:%d", pWrapper->name, pid); - return 0; + dndCloseNodeImp(pWrapper); } static void dndProcessProcHandle(void *handle) { @@ -96,13 +172,14 @@ static void dndProcessProcHandle(void *handle) { static int32_t dndRunInSingleProcess(SDnode *pDnode) { dInfo("dnode run in single process"); + pDnode->procType = PROC_SINGLE; for (EDndType n = DNODE; n < NODE_MAX; ++n) { SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; pWrapper->required = dndRequireNode(pWrapper); if (!pWrapper->required) continue; - if (dndOpenNode(pWrapper) != 0) { + if (dndOpenNodeImp(pWrapper) != 0) { dError("node:%s, failed to start since %s", pWrapper->name, terrstr()); return -1; } @@ -136,8 +213,10 @@ static int32_t dndRunInSingleProcess(SDnode *pDnode) { static int32_t dndRunInParentProcess(SDnode *pDnode) { dInfo("dnode run in parent process"); + pDnode->procType = PROC_PARENT; + SMgmtWrapper *pDWrapper = &pDnode->wrappers[DNODE]; - if (dndOpenNode(pDWrapper) != 0) { + if (dndOpenNodeImp(pDWrapper) != 0) { dError("node:%s, failed to start since %s", pDWrapper->name, terrstr()); return -1; } @@ -146,36 +225,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) { SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; pWrapper->required = dndRequireNode(pWrapper); if (!pWrapper->required) continue; - - int32_t shmsize = tsMnodeShmSize; - if (n == VNODES) { - shmsize = tsVnodeShmSize; - } else if (n == QNODE) { - shmsize = tsQnodeShmSize; - } else if (n == SNODE) { - shmsize = tsSnodeShmSize; - } else if (n == MNODE) { - shmsize = tsMnodeShmSize; - } else if (n == BNODE) { - shmsize = tsBnodeShmSize; - } else { - } - - if (taosCreateShm(&pWrapper->shm, n, shmsize) != 0) { - terrno = TAOS_SYSTEM_ERROR(terrno); - dError("node:%s, failed to create shm size:%d since %s", pWrapper->name, shmsize, terrstr()); - return -1; - } - dInfo("node:%s, shm:%d is created, size:%d", pWrapper->name, pWrapper->shm.id, shmsize); - - SProcCfg cfg = dndGenProcCfg(pWrapper); - cfg.isChild = false; - pWrapper->procType = PROC_PARENT; - pWrapper->pProc = taosProcInit(&cfg); - if (pWrapper->pProc == NULL) { - dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr()); - return -1; - } + if (dndInitNodeProc(pWrapper) != 0) return -1; } if (dndWriteShmFile(pDnode) != 0) { @@ -186,19 +236,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) { for (EDndType n = DNODE + 1; n < NODE_MAX; ++n) { SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; if (!pWrapper->required) continue; - - if (pDnode->ntype == NODE_MAX) { - dInfo("node:%s, should be started manually", pWrapper->name); - } else { - if (dndNewProc(pWrapper, n) != 0) { - return -1; - } - } - - if (taosProcRun(pWrapper->pProc) != 0) { - dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr()); - return -1; - } + if (dndRunNodeProc(pWrapper) != 0) return -1; } dndSetStatus(pDnode, DND_STAT_RUNNING); @@ -239,7 +277,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) { if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) { dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId); taosProcCloseHandles(pWrapper->pProc, dndProcessProcHandle); - dndNewProc(pWrapper, n); + dndNewNodeProc(pWrapper, n); } } } @@ -253,6 +291,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) { static int32_t dndRunInChildProcess(SDnode *pDnode) { SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype]; dInfo("%s run in child process", pWrapper->name); + pDnode->procType = PROC_CHILD; pWrapper->required = dndRequireNode(pWrapper); if (!pWrapper->required) { @@ -264,7 +303,7 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) { tmsgSetDefaultMsgCb(&msgCb); pWrapper->procType = PROC_CHILD; - if (dndOpenNode(pWrapper) != 0) { + if (dndOpenNodeImp(pWrapper) != 0) { dError("node:%s, failed to start since %s", pWrapper->name, terrstr()); return -1; } diff --git a/source/dnode/mgmt/mm/mmHandle.c b/source/dnode/mgmt/mm/mmHandle.c index 6ad0b8c0ed..63240c3224 100644 --- a/source/dnode/mgmt/mm/mmHandle.c +++ b/source/dnode/mgmt/mm/mmHandle.c @@ -80,6 +80,7 @@ int32_t mmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to drop mnode since %s", terrstr()); return -1; } else { + // dndCloseNode(pWrapper); return mmDrop(pWrapper); } } diff --git a/source/dnode/mgmt/qm/qmHandle.c b/source/dnode/mgmt/qm/qmHandle.c index 96fc338529..4fda72759a 100644 --- a/source/dnode/mgmt/qm/qmHandle.c +++ b/source/dnode/mgmt/qm/qmHandle.c @@ -58,7 +58,7 @@ int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to create qnode since %s", terrstr()); return -1; } else { - return qmOpen(pWrapper); + return dndOpenNode(pWrapper); } } @@ -77,6 +77,7 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to drop qnode since %s", terrstr()); return -1; } else { + // dndCloseNode(pWrapper); return qmDrop(pWrapper); } } diff --git a/source/dnode/mgmt/sm/smHandle.c b/source/dnode/mgmt/sm/smHandle.c index 36345cf490..5b30dc04bc 100644 --- a/source/dnode/mgmt/sm/smHandle.c +++ b/source/dnode/mgmt/sm/smHandle.c @@ -58,7 +58,7 @@ int32_t smProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { dError("failed to create snode since %s", terrstr()); return -1; } else { - return smOpen(pWrapper); + return dndOpenNode(pWrapper); } } @@ -78,6 +78,7 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { return -1; } else { return smDrop(pWrapper); + // return dndCloseNode(pWrapper); } } -- GitLab