xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: make more object global

This commit is contained in:
Shengliang Guan 2022-05-16 21:36:29 +08:00
parent 69728a9987
commit 1f886b1117
29 changed files with 247 additions and 316 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsgcb.h
View File

@ -45,7 +45,7 @@ typedef void (*SendRspFp)(const SRpcMsg* pRsp);
typedef void (*SendRedirectRspFp)(const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
typedef void (*RegisterBrokenLinkArgFp)(SMgmtWrapper* pWrapper, SRpcMsg* pMsg);
typedef void (*ReleaseHandleFp)(SMgmtWrapper* pWrapper, void* handle, int8_t type);
typedef void (*ReportStartup)(SMgmtWrapper* pWrapper, const char* name, const char* desc);
typedef void (*ReportStartup)(const char* name, const char* desc);
typedef struct {
SMgmtWrapper* pWrapper;

5
include/dnode/mgmt/dnode.h
View File

@ -40,6 +40,11 @@ void dmCleanup();
*/
int32_t dmRun();
/**
* @brief Stop dnode.
*/
void dmStop();
#ifdef __cplusplus
}
#endif

6
source/common/src/tmsgcb.c
View File

@ -71,9 +71,5 @@ void tmsgReleaseHandle(void* handle, int8_t type) {
void tmsgReportStartup(const char* name, const char* desc) {
ReportStartup fp = tsDefaultMsgCb.reportStartupFp;
if (fp != NULL && tsDefaultMsgCb.pWrapper != NULL) {
(*fp)(tsDefaultMsgCb.pWrapper, name, desc);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
(*fp)(name, desc);
}

74
source/dnode/mgmt/exe/dmMain.c
View File

@ -36,16 +36,10 @@ static struct {
char apolloUrl[PATH_MAX];
const char **envCmd;
SArray *pArgs; // SConfigPair
SDnode *pDnode;
EDndNodeType ntype;
} global = {0};
static void dmStopDnode(int signum, void *info, void *ctx) {
SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
if (pDnode != NULL) {
dmSetEvent(pDnode, DND_EVENT_STOP);
}
}
static void dmStopDnode(int signum, void *info, void *ctx) { dmStop(); }
static void dmSetSignalHandle() {
taosSetSignal(SIGTERM, dmStopDnode);
@ -69,8 +63,8 @@ static void dmSetSignalHandle() {
static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
int32_t cmdEnvIndex = 0;
if (argc < 2) return 0;
global.envCmd = taosMemoryMalloc((argc-1)*sizeof(char*));
memset(global.envCmd, 0, (argc-1)*sizeof(char*));
global.envCmd = taosMemoryMalloc((argc - 1) * sizeof(char *));
memset(global.envCmd, 0, (argc - 1) * sizeof(char *));
for (int32_t i = 1; i < argc; ++i) {
if (strcmp(argv[i], "-c") == 0) {
if (i < argc - 1) {
@ -102,7 +96,8 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
} else if (strcmp(argv[i], "-e") == 0) {
global.envCmd[cmdEnvIndex] = argv[++i];
cmdEnvIndex++;
} else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "--usage") == 0 || strcmp(argv[i], "-?")) {
} else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "--usage") == 0 ||
strcmp(argv[i], "-?")) {
global.printHelp = true;
} else {
}
@ -144,23 +139,6 @@ static void dmDumpCfg() {
cfgDumpCfg(pCfg, 0, true);
}
static SDnodeOpt dmGetOpt() {
SConfig *pCfg = taosGetCfg();
SDnodeOpt option = {0};
option.numOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
tstrncpy(option.dataDir, tsDataDir, sizeof(option.dataDir));
tstrncpy(option.firstEp, tsFirst, sizeof(option.firstEp));
tstrncpy(option.secondEp, tsSecond, sizeof(option.firstEp));
option.serverPort = tsServerPort;
tstrncpy(option.localFqdn, tsLocalFqdn, sizeof(option.localFqdn));
snprintf(option.localEp, sizeof(option.localEp), "%s:%u", option.localFqdn, option.serverPort);
option.disks = tsDiskCfg;
option.numOfDisks = tsDiskCfgNum;
option.ntype = global.ntype;
return option;
}
static int32_t dmInitLog() {
char logName[12] = {0};
snprintf(logName, sizeof(logName), "%slog", dmNodeLogName(global.ntype));
@ -175,34 +153,6 @@ static void dmSetProcInfo(int32_t argc, char **argv) {
}
}
static int32_t dmRunDnode() {
if (dmInit() != 0) {
dError("failed to init environment since %s", terrstr());
return -1;
}
SDnodeOpt option = dmGetOpt();
SDnode *pDnode = dmCreate(&option);
if (pDnode == NULL) {
dError("failed to to create dnode since %s", terrstr());
return -1;
} else {
global.pDnode = pDnode;
dmSetSignalHandle();
}
dInfo("start to run dnode");
int32_t code = dmRun(pDnode);
dInfo("shutting down the service");
global.pDnode = NULL;
dmClose(pDnode);
dmCleanup();
taosCloseLog();
taosCleanupCfg();
return code;
}
static void taosCleanupArgs() {
if (global.envCmd != NULL) taosMemoryFree(global.envCmd);
}
@ -259,5 +209,17 @@ int main(int argc, char const *argv[]) {
dmSetProcInfo(argc, (char **)argv);
taosCleanupArgs();
return dmRunDnode();
if (dmInit(global.ntype) != 0) {
dError("failed to init dnode since %s", terrstr());
return -1;
}
dInfo("start to run dnode");
dmSetSignalHandle();
int32_t code = dmRun();
dInfo("shutting down the service");
dmCleanup();
return code;
}

2
source/dnode/mgmt/mgmt_bnode/inc/bmInt.h
View File

@ -25,11 +25,11 @@ extern "C" {
#endif
typedef struct SBnodeMgmt {
SDnodeData *pData;
SBnode *pBnode;
SMsgCb msgCb;
const char *path;
const char *name;
int32_t dnodeId;
SMultiWorker writeWorker;
SSingleWorker monitorWorker;
} SBnodeMgmt;

2
source/dnode/mgmt/mgmt_bnode/src/bmHandle.c
View File

@ -76,7 +76,7 @@ int32_t bmProcessDropReq(SBnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
if (pMgmt->dnodeId != 0 && dropReq.dnodeId != pMgmt->dnodeId) {
if (pMgmt->pData->dnodeId != 0 && dropReq.dnodeId != pMgmt->pData->dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop bnode since %s", terrstr());
return -1;

2
source/dnode/mgmt/mgmt_bnode/src/bmInt.c
View File

@ -39,9 +39,9 @@ int32_t bmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
return -1;
}
pMgmt->pData = pInput->pData;
pMgmt->path = pInput->path;
pMgmt->name = pInput->name;
pMgmt->dnodeId = pInput->pData->dnodeId;
pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.pMgmt = pMgmt;

1
source/dnode/mgmt/mgmt_dnode/inc/dmInt.h
View File

@ -23,7 +23,6 @@ extern "C" {
#endif
typedef struct SDnodeMgmt {
struct SDnode *pDnode;
SDnodeData *pData;
SMsgCb msgCb;
const char *path;

4
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
View File

@ -59,8 +59,8 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
req.rebootTime = pMgmt->pData->rebootTime;
req.updateTime = pMgmt->pData->updateTime;
req.numOfCores = tsNumOfCores;
req.numOfSupportVnodes = pMgmt->pData->supportVnodes;
tstrncpy(req.dnodeEp, pMgmt->pData->localEp, TSDB_EP_LEN);
req.numOfSupportVnodes = tsNumOfSupportVnodes;
tstrncpy(req.dnodeEp, tsLocalEp, TSDB_EP_LEN);
req.clusterCfg.statusInterval = tsStatusInterval;
req.clusterCfg.checkTime = 0;

1
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
View File

@ -39,7 +39,6 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
return -1;
}
pMgmt->pDnode = pInput->pDnode;
pMgmt->pData = pInput->pData;
pMgmt->msgCb = pInput->msgCb;
pMgmt->path = pInput->path;

33
source/dnode/mgmt/mgmt_dnode/src/dmMonitor.c
View File

@ -16,19 +16,18 @@
#define _DEFAULT_SOURCE
#include "dmInt.h"
#define dmSendLocalRecv(pMgmt, mtype, func, pInfo) \
if (!tsMultiProcess) { \
SRpcMsg rsp = {0}; \
SRpcMsg req = {.msgType = mtype}; \
SEpSet epset = {.inUse = 0, .numOfEps = 1}; \
tstrncpy(epset.eps[0].fqdn, pMgmt->pData->localFqdn, TSDB_FQDN_LEN); \
epset.eps[0].port = pMgmt->pData->serverPort; \
\
rpcSendRecv(pMgmt->msgCb.clientRpc, &epset, &req, &rsp); \
if (rsp.code == 0 && rsp.contLen > 0) { \
func(rsp.pCont, rsp.contLen, pInfo); \
} \
rpcFreeCont(rsp.pCont); \
#define dmSendLocalRecv(pMgmt, mtype, func, pInfo) \
if (!tsMultiProcess) { \
SRpcMsg rsp = {0}; \
SRpcMsg req = {.msgType = mtype}; \
SEpSet epset = {.inUse = 0, .numOfEps = 1}; \
tstrncpy(epset.eps[0].fqdn, tsLocalFqdn, TSDB_FQDN_LEN); \
epset.eps[0].port = tsServerPort; \
rpcSendRecv(pMgmt->msgCb.clientRpc, &epset, &req, &rsp); \
if (rsp.code == 0 && rsp.contLen > 0) { \
func(rsp.pCont, rsp.contLen, pInfo); \
} \
rpcFreeCont(rsp.pCont); \
}
static void dmGetMonitorBasicInfo(SDnodeMgmt *pMgmt, SMonBasicInfo *pInfo) {
@ -40,10 +39,10 @@ static void dmGetMonitorBasicInfo(SDnodeMgmt *pMgmt, SMonBasicInfo *pInfo) {
static void dmGetMonitorDnodeInfo(SDnodeMgmt *pMgmt, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pMgmt->pData->rebootTime) / (86400000.0f);
pInfo->has_mnode = (*pMgmt->isNodeRequiredFp)(pMgmt->pDnode, MNODE);
pInfo->has_qnode = (*pMgmt->isNodeRequiredFp)(pMgmt->pDnode, QNODE);
pInfo->has_snode = (*pMgmt->isNodeRequiredFp)(pMgmt->pDnode, SNODE);
pInfo->has_bnode = (*pMgmt->isNodeRequiredFp)(pMgmt->pDnode, BNODE);
pInfo->has_mnode = (*pMgmt->isNodeRequiredFp)(MNODE);
pInfo->has_qnode = (*pMgmt->isNodeRequiredFp)(QNODE);
pInfo->has_snode = (*pMgmt->isNodeRequiredFp)(SNODE);
pInfo->has_bnode = (*pMgmt->isNodeRequiredFp)(BNODE);
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));

16
source/dnode/mgmt/mgmt_dnode/src/dmWorker.c
View File

@ -116,28 +116,28 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
code = dmProcessGrantRsp(pMgmt, pMsg);
break;
case TDMT_DND_CREATE_MNODE:
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, MNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(MNODE, pMsg);
break;
case TDMT_DND_DROP_MNODE:
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, MNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(MNODE, pMsg);
break;
case TDMT_DND_CREATE_QNODE:
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, QNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(QNODE, pMsg);
break;
case TDMT_DND_DROP_QNODE:
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, QNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(QNODE, pMsg);
break;
case TDMT_DND_CREATE_SNODE:
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, SNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(SNODE, pMsg);
break;
case TDMT_DND_DROP_SNODE:
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, SNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(SNODE, pMsg);
break;
case TDMT_DND_CREATE_BNODE:
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, BNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(BNODE, pMsg);
break;
case TDMT_DND_DROP_BNODE:
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, BNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(BNODE, pMsg);
break;
case TDMT_DND_SERVER_STATUS:
code = dmProcessServerRunStatus(pMgmt, pMsg);

2
source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
View File

@ -24,11 +24,11 @@ extern "C" {
#endif
typedef struct SMnodeMgmt {
SDnodeData *pData;
SMnode *pMnode;
SMsgCb msgCb;
const char *path;
const char *name;
int32_t dnodeId;
SSingleWorker queryWorker;
SSingleWorker readWorker;
SSingleWorker writeWorker;

6
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
View File

@ -109,7 +109,7 @@ int32_t mmProcessDropReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
if (pMgmt->dnodeId != 0 && dropReq.dnodeId != pMgmt->dnodeId) {
if (pMgmt->pData->dnodeId != 0 && dropReq.dnodeId != pMgmt->pData->dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop mnode since %s", terrstr());
return -1;
@ -133,9 +133,9 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
if (pMgmt->dnodeId != 0 && alterReq.dnodeId != pMgmt->dnodeId) {
if (pMgmt->pData->dnodeId != 0 && alterReq.dnodeId != pMgmt->pData->dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to alter mnode since %s, input:%d cur:%d", terrstr(), alterReq.dnodeId, pMgmt->dnodeId);
dError("failed to alter mnode since %s, input:%d cur:%d", terrstr(), alterReq.dnodeId, pMgmt->pData->dnodeId);
return -1;
} else {
return mmAlter(pMgmt, &alterReq);

8
source/dnode/mgmt/mgmt_mnode/src/mmInt.c
View File

@ -70,7 +70,7 @@ static int32_t mmBuildOptionFromReq(SMnodeMgmt *pMgmt, SMnodeOpt *pOption, SDCre
pReplica->id = pCreate->replicas[i].id;
pReplica->port = pCreate->replicas[i].port;
memcpy(pReplica->fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
if (pReplica->id == pMgmt->dnodeId) {
if (pReplica->id == pMgmt->pData->dnodeId) {
pOption->selfIndex = i;
}
}
@ -128,9 +128,9 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
return -1;
}
pMgmt->pData = pInput->pData;
pMgmt->path = pInput->path;
pMgmt->name = pInput->name;
pMgmt->dnodeId = pInput->pData->dnodeId;
pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)mmPutRpcMsgToQueryQueue;
pMgmt->msgCb.queueFps[READ_QUEUE] = (PutToQueueFp)mmPutRpcMsgToReadQueue;
@ -148,7 +148,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
SMnodeOpt option = {0};
if (!deployed) {
dInfo("mnode start to deploy");
pMgmt->dnodeId = 1;
pMgmt->pData->dnodeId = 1;
mmBuildOptionForDeploy(pMgmt, pInput, &option);
} else {
dInfo("mnode start to open");
@ -178,7 +178,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
}
}
pInput->pData->dnodeId = pMgmt->dnodeId;
pInput->pData->dnodeId = pMgmt->pData->dnodeId;
pOutput->pMgmt = pMgmt;
return 0;
}

2
source/dnode/mgmt/mgmt_qnode/inc/qmInt.h
View File

@ -25,11 +25,11 @@ extern "C" {
#endif
typedef struct SQnodeMgmt {
SDnodeData *pData;
SQnode *pQnode;
SMsgCb msgCb;
const char *path;
const char *name;
int32_t dnodeId;
SSingleWorker queryWorker;
SSingleWorker fetchWorker;
SSingleWorker monitorWorker;

2
source/dnode/mgmt/mgmt_qnode/src/qmHandle.c
View File

@ -76,7 +76,7 @@ int32_t qmProcessDropReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
if (pMgmt->dnodeId != 0 && dropReq.dnodeId != pMgmt->dnodeId) {
if (pMgmt->pData->dnodeId != 0 && dropReq.dnodeId != pMgmt->pData->dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop qnode since %s", terrstr());
return -1;

2
source/dnode/mgmt/mgmt_qnode/src/qmInt.c
View File

@ -39,9 +39,9 @@ static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
return -1;
}
pMgmt->pData = pInput->pData;
pMgmt->path = pInput->path;
pMgmt->name = pInput->name;
pMgmt->dnodeId = pInput->pData->dnodeId;
pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qmPutRpcMsgToQueryQueue;
pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)qmPutRpcMsgToFetchQueue;

2
source/dnode/mgmt/mgmt_snode/inc/smInt.h
View File

@ -25,11 +25,11 @@ extern "C" {
#endif
typedef struct SSnodeMgmt {
SDnodeData *pData;
SSnode *pSnode;
SMsgCb msgCb;
const char *path;
const char *name;
int32_t dnodeId;
SRWLatch latch;
int8_t uniqueWorkerInUse;
SArray *uniqueWorkers; // SArray<SMultiWorker*>

2
source/dnode/mgmt/mgmt_snode/src/smHandle.c
View File

@ -76,7 +76,7 @@ int32_t smProcessDropReq(SSnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
if (pMgmt->dnodeId != 0 && dropReq.dnodeId != pMgmt->dnodeId) {
if (pMgmt->pData->dnodeId != 0 && dropReq.dnodeId != pMgmt->pData->dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop snode since %s", terrstr());
return -1;

2
source/dnode/mgmt/mgmt_snode/src/smInt.c
View File

@ -40,9 +40,9 @@ int32_t smOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
return -1;
}
pMgmt->pData = pInput->pData;
pMgmt->path = pInput->path;
pMgmt->name = pInput->name;
pMgmt->dnodeId = pInput->pData->dnodeId;
pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.pMgmt = pMgmt;

2
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -26,10 +26,10 @@ extern "C" {
#endif
typedef struct SVnodeMgmt {
SDnodeData *pData;
SMsgCb msgCb;
const char *path;
const char *name;
int32_t dnodeId;
SQWorkerPool queryPool;
SQWorkerPool fetchPool;
SWWorkerPool syncPool;

2
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -247,9 +247,9 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
SVnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SVnodeMgmt));
if (pMgmt == NULL) goto _OVER;
pMgmt->pData = pInput->pData;
pMgmt->path = pInput->path;
pMgmt->name = pInput->name;
pMgmt->dnodeId = pInput->pData->dnodeId;
pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToWriteQueue;
pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)vmPutRpcMsgToSyncQueue;

52
source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
View File

@ -65,29 +65,29 @@ typedef struct {
} SProc;
typedef struct SMgmtWrapper {
SDnode *pDnode;
SMgmtFunc func;
void *pMgmt;
const char *name;
char *path;
int32_t refCount;
SRWLatch latch;
EDndNodeType ntype;
bool deployed;
bool required;
SProc proc;
NodeMsgFp msgFps[TDMT_MAX];
struct SDnode *pDnode;
SMgmtFunc func;
void *pMgmt;
const char *name;
char *path;
int32_t refCount;
SRWLatch latch;
EDndNodeType ntype;
bool deployed;
bool required;
SProc proc;
NodeMsgFp msgFps[TDMT_MAX];
} SMgmtWrapper;
typedef struct {
EDndNodeType defaultNtype;
bool needCheckVgId;
} SMsgHandle;
} SDnodeHandle;
typedef struct {
void *serverRpc;
void *clientRpc;
SMsgHandle msgHandles[TDMT_MAX];
void *serverRpc;
void *clientRpc;
SDnodeHandle msgHandles[TDMT_MAX];
} SDnodeTrans;
typedef struct {
@ -110,9 +110,10 @@ typedef struct SUdfdData {
} SUdfdData;
typedef struct SDnode {
int8_t once;
bool stop;
EDndProcType ptype;
EDndNodeType rtype;
EDndEvent event;
EDndRunStatus status;
SStartupInfo startup;
SDnodeTrans trans;
@ -123,23 +124,26 @@ typedef struct SDnode {
SMgmtWrapper wrappers[NODE_END];
} SDnode;
// dmExec.c
int32_t dmOpenNode(SMgmtWrapper *pWrapper);
void dmCloseNode(SMgmtWrapper *pWrapper);
// dmEmv.c
void dmReportStartup(const char *pName, const char *pDesc);
// dmObj.c
// dmMgmt.c
int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype);
void dmCleanupDnode(SDnode *pDnode);
SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType nType);
int32_t dmMarkWrapper(SMgmtWrapper *pWrapper);
void dmReleaseWrapper(SMgmtWrapper *pWrapper);
SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper);
void dmSetStatus(SDnode *pDnode, EDndRunStatus stype);
void dmSetEvent(SDnode *pDnode, EDndEvent event);
void dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc);
void dmProcessServerStartupStatus(SDnode *pDnode, SRpcMsg *pMsg);
void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pMsg);
// dmNodes.c
int32_t dmOpenNode(SMgmtWrapper *pWrapper);
void dmCloseNode(SMgmtWrapper *pWrapper);
int32_t dmRunDnode(SDnode *pDnode);
// dmProc.c
int32_t dmInitProc(struct SMgmtWrapper *pWrapper);
void dmCleanupProc(struct SMgmtWrapper *pWrapper);

138
source/dnode/mgmt/node_mgmt/src/dmEnv.c
View File

@ -16,23 +16,10 @@
#define _DEFAULT_SOURCE
#include "dmMgmt.h"
static struct {
int8_t once;
EDndProcType ptype;
EDndNodeType rtype;
EDndEvent event;
EDndRunStatus status;
SStartupInfo startup;
SDnodeTrans trans;
SUdfdData udfdData;
TdThreadMutex mutex;
TdFilePtr lockfile;
SDnodeData data;
SMgmtWrapper wrappers[NODE_END];
} global;
static SDnode global = {0};
static int32_t dmCheckRepeatInit() {
if (atomic_val_compare_exchange_8(&global.once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
static int32_t dmCheckRepeatInit(SDnode *pDnode) {
if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
dError("env is already initialized");
terrno = TSDB_CODE_REPEAT_INIT;
return -1;
@ -60,19 +47,19 @@ static int32_t dmInitMonitor() {
return 0;
}
int32_t dmInit() {
int32_t dmInit(int8_t rtype) {
dInfo("start to init env");
if (dmCheckRepeatInit() != 0) return -1;
if (dmCheckRepeatInit(&global) != 0) return -1;
if (dmInitSystem() != 0) return -1;
if (dmInitMonitor() != 0) return -1;
// if (dmInit)
if (dmInitDnode(&global, rtype) != 0) return -1;
dInfo("env is initialized");
return 0;
}
static int32_t dmCheckRepeatCleanup() {
if (atomic_val_compare_exchange_8(&global.once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
static int32_t dmCheckRepeatCleanup(SDnode *pDnode) {
if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("env is already cleaned up");
return -1;
}
@ -82,7 +69,7 @@ static int32_t dmCheckRepeatCleanup() {
void dmCleanup() {
dDebug("start to cleanup env");
if (dmCheckRepeatCleanup != 0) return;
dmCleanupDnode(&global);
monCleanup();
syncCleanUp();
walCleanUp();
@ -90,4 +77,111 @@ void dmCleanup() {
udfStopUdfd();
taosStopCacheRefreshWorker();
dInfo("env is cleaned up");
taosCloseLog();
taosCleanupCfg();
}
void dmStop() {
SDnode *pDnode = &global;
pDnode->stop = true;
}
int32_t dmRun() {
SDnode *pDnode = &global;
return dmRunDnode(pDnode);
}
static int32_t dmProcessCreateNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
SDnode *pDnode = &global;
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
dmReleaseWrapper(pWrapper);
terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
dError("failed to create node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
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;
}
SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);
int32_t code = (*pWrapper->func.createFp)(&input, pMsg);
if (code != 0) {
dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been created", pWrapper->name);
(void)dmOpenNode(pWrapper);
pWrapper->required = true;
pWrapper->deployed = true;
pWrapper->proc.ptype = pDnode->ptype;
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
static int32_t dmProcessDropNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
SDnode *pDnode = &global;
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
dError("failed to drop node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
int32_t code = (*pWrapper->func.dropFp)(pWrapper->pMgmt, pMsg);
if (code != 0) {
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been dropped", pWrapper->name);
pWrapper->required = false;
pWrapper->deployed = false;
}
dmReleaseWrapper(pWrapper);
if (code == 0) {
dmCloseNode(pWrapper);
taosRemoveDir(pWrapper->path);
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
static bool dmIsNodeRequired(EDndNodeType ntype) {
SDnode *pDnode = &global;
return pDnode->wrappers[ntype].required;
}
SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) {
SMgmtInputOpt opt = {
.path = pWrapper->path,
.name = pWrapper->name,
.pData = &pWrapper->pDnode->data,
.processCreateNodeFp = dmProcessCreateNodeReq,
.processDropNodeFp = dmProcessDropNodeReq,
.isNodeRequiredFp = dmIsNodeRequired,
};
opt.msgCb = dmGetMsgcb(pWrapper);
return opt;
}
void dmReportStartup(const char *pName, const char *pDesc) {
SStartupInfo *pStartup = &global.startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
dDebug("step:%s, %s", pStartup->name, pStartup->desc);
}

147
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
View File

@ -16,8 +16,6 @@
#define _DEFAULT_SOURCE
#include "dmMgmt.h"
static bool dmIsNodeRequired(SDnode *pDnode, EDndNodeType ntype) { return pDnode->wrappers[ntype].required; }
static bool dmRequireNode(SMgmtWrapper *pWrapper) {
SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);
@ -39,8 +37,8 @@ static bool dmRequireNode(SMgmtWrapper *pWrapper) {
return required;
}
static int32_t dmInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->rtype = pOption->ntype;
static int32_t dmInitVars(SDnode *pDnode, EDndNodeType rtype) {
pDnode->rtype = rtype;
if (tsMultiProcess == 0) {
pDnode->ptype = DND_PROC_SINGLE;
@ -65,21 +63,6 @@ static int32_t dmInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pData->rebootTime = taosGetTimestampMs();
pData->dropped = 0;
pData->stopped = 0;
pData->localEp = strdup(pOption->localEp);
pData->localFqdn = strdup(pOption->localFqdn);
pData->firstEp = strdup(pOption->firstEp);
pData->secondEp = strdup(pOption->secondEp);
pData->supportVnodes = pOption->numOfSupportVnodes;
pData->serverPort = pOption->serverPort;
pData->numOfDisks = pOption->numOfDisks;
pData->disks = pOption->disks;
pData->dataDir = strdup(pOption->dataDir);
if (pData->dataDir == NULL || pData->localEp == NULL || pData->localFqdn == NULL || pData->firstEp == NULL ||
pData->secondEp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pData->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
if (pData->dnodeHash == NULL) {
@ -126,30 +109,17 @@ static void dmClearVars(SDnode *pDnode) {
}
taosWUnLockLatch(&pData->latch);
taosMemoryFreeClear(pData->localEp);
taosMemoryFreeClear(pData->localFqdn);
taosMemoryFreeClear(pData->firstEp);
taosMemoryFreeClear(pData->secondEp);
taosMemoryFreeClear(pData->dataDir);
taosThreadMutexDestroy(&pDnode->mutex);
memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
taosMemoryFree(pDnode);
}
SDnode *dmCreate(const SDnodeOpt *pOption) {
int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {
dInfo("start to create dnode");
int32_t code = -1;
char path[PATH_MAX + 100] = {0};
SDnode *pDnode = NULL;
pDnode = taosMemoryCalloc(1, sizeof(SDnode));
if (pDnode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
if (dmInitVars(pDnode, pOption) != 0) {
if (dmInitVars(pDnode, rtype) != 0) {
goto _OVER;
}
@ -162,7 +132,6 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
pWrapper->pDnode = pDnode;
pWrapper->name = dmNodeName(ntype);
pWrapper->ntype = ntype;
pWrapper->proc.wrapper = pWrapper;
@ -174,7 +143,7 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
}
taosInitRWLatch(&pWrapper->latch);
snprintf(path, sizeof(path), "%s%s%s", pOption->dataDir, TD_DIRSEP, pWrapper->name);
snprintf(path, sizeof(path), "%s%s%s", tsDataDir, TD_DIRSEP, pWrapper->name);
pWrapper->path = strdup(path);
if (pWrapper->path == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -195,7 +164,7 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
}
if (OnlyInSingleProc(pDnode->ptype) || InParentProc(pDnode->ptype)) {
pDnode->lockfile = dmCheckRunning(pOption->dataDir);
pDnode->lockfile = dmCheckRunning(tsDataDir);
if (pDnode->lockfile == NULL) {
goto _OVER;
}
@ -210,7 +179,7 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
goto _OVER;
}
dmReportStartup(pDnode, "dnode-transport", "initialized");
dmReportStartup("dnode-transport", "initialized");
dInfo("dnode is created, ptr:%p", pDnode);
code = 0;
@ -221,10 +190,10 @@ _OVER:
dError("failed to create dnode since %s", terrstr());
}
return pDnode;
return code;
}
void dmClose(SDnode *pDnode) {
void dmCleanupDnode(SDnode *pDnode) {
if (pDnode == NULL) return;
dmCleanupClient(pDnode);
@ -240,12 +209,6 @@ void dmSetStatus(SDnode *pDnode, EDndRunStatus status) {
}
}
void dmSetEvent(SDnode *pDnode, EDndEvent event) {
if (event == DND_EVENT_STOP) {
pDnode->event = event;
}
}
SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SMgmtWrapper *pRetWrapper = pWrapper;
@ -288,17 +251,6 @@ void dmReleaseWrapper(SMgmtWrapper *pWrapper) {
dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
}
void dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupInfo *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
dDebug("step:%s, %s", pStartup->name, pStartup->desc);
}
void dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc) {
dmReportStartup(pWrapper->pDnode, pName, pDesc);
}
static void dmGetServerStartupStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
SDnodeMgmt *pMgmt = pDnode->wrappers[DNODE].pMgmt;
pStatus->details[0] = 0;
@ -315,7 +267,7 @@ static void dmGetServerStartupStatus(SDnode *pDnode, SServerStatusRsp *pStatus)
void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("net test req is received");
SRpcMsg rsp = {.info = pReq->info, .code = 0};
SRpcMsg rsp = {.code = 0, .info = pReq->info};
rsp.pCont = rpcMallocCont(pReq->contLen);
if (rsp.pCont == NULL) {
rsp.code = TSDB_CODE_OUT_OF_MEMORY;
@ -353,82 +305,3 @@ _OVER:
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
}
static int32_t dmProcessCreateNodeReq(SDnode *pDnode, EDndNodeType ntype, SRpcMsg *pMsg) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
dmReleaseWrapper(pWrapper);
terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
dError("failed to create node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
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;
}
SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);
int32_t code = (*pWrapper->func.createFp)(&input, pMsg);
if (code != 0) {
dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been created", pWrapper->name);
(void)dmOpenNode(pWrapper);
pWrapper->required = true;
pWrapper->deployed = true;
pWrapper->proc.ptype = pDnode->ptype;
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
static int32_t dmProcessDropNodeReq(SDnode *pDnode, EDndNodeType ntype, SRpcMsg *pMsg) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
dError("failed to drop node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
int32_t code = (*pWrapper->func.dropFp)(pWrapper->pMgmt, pMsg);
if (code != 0) {
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been dropped", pWrapper->name);
pWrapper->required = false;
pWrapper->deployed = false;
}
dmReleaseWrapper(pWrapper);
if (code == 0) {
dmCloseNode(pWrapper);
taosRemoveDir(pWrapper->path);
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) {
SMgmtInputOpt opt = {
.pDnode = pWrapper->pDnode,
.pData = &pWrapper->pDnode->data,
.processCreateNodeFp = dmProcessCreateNodeReq,
.processDropNodeFp = dmProcessDropNodeReq,
.isNodeRequiredFp = dmIsNodeRequired,
.name = pWrapper->name,
.path = pWrapper->path,
};
opt.msgCb = dmGetMsgcb(pWrapper);
return opt;
}

10
source/dnode/mgmt/node_mgmt/src/dmNodes.c
View File

@ -136,7 +136,7 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
pWrapper->pMgmt = output.pMgmt;
}
dmReportStartup(pWrapper->pDnode, pWrapper->name, "openned");
dmReportStartup(pWrapper->name, "openned");
return 0;
}
@ -151,7 +151,7 @@ int32_t dmStartNode(SMgmtWrapper *pWrapper) {
dDebug("node:%s, has been started", pWrapper->name);
}
dmReportStartup(pWrapper->pDnode, pWrapper->name, "started");
dmReportStartup(pWrapper->name, "started");
return 0;
}
@ -221,7 +221,7 @@ static int32_t dmStartNodes(SDnode *pDnode) {
}
dInfo("TDengine initialized successfully");
dmReportStartup(pDnode, "TDengine", "initialized successfully");
dmReportStartup("TDengine", "initialized successfully");
return 0;
}
@ -260,7 +260,7 @@ static void dmWatchNodes(SDnode *pDnode) {
taosThreadMutexUnlock(&pDnode->mutex);
}
int32_t dmRun(SDnode *pDnode) {
int32_t dnRunDnode(SDnode *pDnode) {
if (dmOpenNodes(pDnode) != 0) {
dError("failed to open nodes since %s", terrstr());
return -1;
@ -272,7 +272,7 @@ int32_t dmRun(SDnode *pDnode) {
}
while (1) {
if (pDnode->event & DND_EVENT_STOP) {
if (!pDnode->stop) {
dInfo("dnode is about to stop");
dmSetStatus(pDnode, DND_STAT_STOPPED);
dmStopNodes(pDnode);

12
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -52,7 +52,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
int32_t code = -1;
SRpcMsg *pMsg = NULL;
bool needRelease = false;
SMsgHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)];
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)];
SMgmtWrapper *pWrapper = NULL;
dTrace("msg:%s is received, handle:%p cont:%p len:%d code:0x%04x app:%p refId:%" PRId64, TMSG_INFO(pRpc->msgType),
@ -178,7 +178,7 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SMgmtHandle *pMgmt = taosArrayGet(pArray, i);
SMsgHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
if (pMgmt->needCheckVgId) {
pHandle->needCheckVgId = pMgmt->needCheckVgId;
}
@ -201,7 +201,7 @@ static void dmSendRpcRedirectRsp(SDnode *pDnode, const SRpcMsg *pReq) {
dDebug("RPC %p, req is redirected, num:%d use:%d", pReq->info.handle, epSet.numOfEps, epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) {
dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
if (strcmp(epSet.eps[i].fqdn, pDnode->data.localFqdn) == 0 && epSet.eps[i].port == pDnode->data.serverPort) {
if (strcmp(epSet.eps[i].fqdn, tsLocalFqdn) == 0 && epSet.eps[i].port == tsServerPort) {
epSet.inUse = (i + 1) % epSet.numOfEps;
}
@ -410,8 +410,8 @@ int32_t dmInitServer(SDnode *pDnode) {
SRpcInit rpcInit = {0};
strncpy(rpcInit.localFqdn, pDnode->data.localFqdn, strlen(pDnode->data.localFqdn));
rpcInit.localPort = pDnode->data.serverPort;
strncpy(rpcInit.localFqdn, tsLocalFqdn, strlen(tsLocalFqdn));
rpcInit.localPort = tsServerPort;
rpcInit.label = "DND";
rpcInit.numOfThreads = tsNumOfRpcThreads;
rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
@ -449,7 +449,7 @@ SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper) {
.sendRedirectRspFp = dmSendRedirectRsp,
.registerBrokenLinkArgFp = dmRegisterBrokenLinkArg,
.releaseHandleFp = dmReleaseHandle,
.reportStartupFp = dmReportStartupByWrapper,
.reportStartupFp = dmReportStartup,
};
return msgCb;
}

24
source/dnode/mgmt/node_util/inc/dmUtil.h
View File

@ -92,18 +92,18 @@ typedef int32_t (*ProcessDropNodeFp)(EDndNodeType ntype, SRpcMsg *pMsg);
typedef bool (*IsNodeRequiredFp)(EDndNodeType ntype);
typedef struct {
int32_t dnodeId;
int64_t clusterId;
int64_t dnodeVer;
int64_t updateTime;
int64_t rebootTime;
bool dropped;
bool stopped;
SEpSet mnodeEps;
SArray *dnodeEps;
SHashObj *dnodeHash;
SRWLatch latch;
SMsgCb msgCb;
int32_t dnodeId;
int64_t clusterId;
int64_t dnodeVer;
int64_t updateTime;
int64_t rebootTime;
bool dropped;
bool stopped;
SEpSet mnodeEps;
SArray *dnodeEps;
SHashObj *dnodeHash;
SRWLatch latch;
SMsgCb msgCb;
} SDnodeData;
typedef struct {