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

refact(cluster): node mgmt

This commit is contained in:
Shengliang Guan 2022-04-12 15:49:21 +08:00
parent 0094f30140
commit 4fedc23b26
31 changed files with 628 additions and 538 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/tmsg.h
View File

@ -744,8 +744,8 @@ typedef struct {
} SVnodeLoad; } SVnodeLoad;
typedef struct { typedef struct {
int32_t sver; // software version int32_t sver; // software version
int64_t dver; // dnode table version in sdb int64_t dnodeVer; // dnode table version in sdb
int32_t dnodeId; int32_t dnodeId;
int64_t clusterId; int64_t clusterId;
int64_t rebootTime; int64_t rebootTime;
@ -772,7 +772,7 @@ typedef struct {
} SDnodeEp; } SDnodeEp;
typedef struct { typedef struct {
int64_t dver; int64_t dnodeVer;
SDnodeCfg dnodeCfg; SDnodeCfg dnodeCfg;
SArray* pDnodeEps; // Array of SDnodeEp SArray* pDnodeEps; // Array of SDnodeEp
} SStatusRsp; } SStatusRsp;

8
source/common/src/tmsg.c
View File

@ -868,7 +868,7 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
// status // status
if (tEncodeI32(&encoder, pReq->sver) < 0) return -1; if (tEncodeI32(&encoder, pReq->sver) < 0) return -1;
if (tEncodeI64(&encoder, pReq->dver) < 0) return -1; if (tEncodeI64(&encoder, pReq->dnodeVer) < 0) return -1;
if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1; if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
if (tEncodeI64(&encoder, pReq->clusterId) < 0) return -1; if (tEncodeI64(&encoder, pReq->clusterId) < 0) return -1;
if (tEncodeI64(&encoder, pReq->rebootTime) < 0) return -1; if (tEncodeI64(&encoder, pReq->rebootTime) < 0) return -1;
@ -913,7 +913,7 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
// status // status
if (tDecodeI32(&decoder, &pReq->sver) < 0) return -1; if (tDecodeI32(&decoder, &pReq->sver) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->dver) < 0) return -1; if (tDecodeI64(&decoder, &pReq->dnodeVer) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1; if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->clusterId) < 0) return -1; if (tDecodeI64(&decoder, &pReq->clusterId) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->rebootTime) < 0) return -1; if (tDecodeI64(&decoder, &pReq->rebootTime) < 0) return -1;
@ -965,7 +965,7 @@ int32_t tSerializeSStatusRsp(void *buf, int32_t bufLen, SStatusRsp *pRsp) {
if (tStartEncode(&encoder) < 0) return -1; if (tStartEncode(&encoder) < 0) return -1;
// status // status
if (tEncodeI64(&encoder, pRsp->dver) < 0) return -1; if (tEncodeI64(&encoder, pRsp->dnodeVer) < 0) return -1;
// dnode cfg // dnode cfg
if (tEncodeI32(&encoder, pRsp->dnodeCfg.dnodeId) < 0) return -1; if (tEncodeI32(&encoder, pRsp->dnodeCfg.dnodeId) < 0) return -1;
@ -996,7 +996,7 @@ int32_t tDeserializeSStatusRsp(void *buf, int32_t bufLen, SStatusRsp *pRsp) {
if (tStartDecode(&decoder) < 0) return -1; if (tStartDecode(&decoder) < 0) return -1;
// status // status
if (tDecodeI64(&decoder, &pRsp->dver) < 0) return -1; if (tDecodeI64(&decoder, &pRsp->dnodeVer) < 0) return -1;
// cluster cfg // cluster cfg
if (tDecodeI32(&decoder, &pRsp->dnodeCfg.dnodeId) < 0) return -1; if (tDecodeI32(&decoder, &pRsp->dnodeCfg.dnodeId) < 0) return -1;

7
source/dnode/mgmt/CMakeLists.txt
View File

@ -1,3 +1,5 @@
add_subdirectory(interface)
aux_source_directory(dm DNODE_SRC) aux_source_directory(dm DNODE_SRC)
aux_source_directory(qm DNODE_SRC) aux_source_directory(qm DNODE_SRC)
aux_source_directory(bm DNODE_SRC) aux_source_directory(bm DNODE_SRC)
@ -7,19 +9,20 @@ aux_source_directory(mm DNODE_SRC)
aux_source_directory(main DNODE_SRC) aux_source_directory(main DNODE_SRC)
add_library(dnode STATIC ${DNODE_SRC}) add_library(dnode STATIC ${DNODE_SRC})
target_link_libraries( target_link_libraries(
dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor dnode dnode_interface
) )
target_include_directories( target_include_directories(
dnode dnode
PUBLIC "${TD_SOURCE_DIR}/include/dnode/mgmt" PUBLIC "${TD_SOURCE_DIR}/include/dnode/mgmt"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc" PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/interface/inc"
) )
aux_source_directory(exe EXEC_SRC) aux_source_directory(exe EXEC_SRC)
add_executable(taosd ${EXEC_SRC}) add_executable(taosd ${EXEC_SRC})
target_include_directories( target_include_directories(
taosd taosd
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc" PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/interface/inc"
) )
target_link_libraries(taosd dnode) target_link_libraries(taosd dnode)

6
source/dnode/mgmt/bm/bmHandle.c
View File

@ -53,9 +53,9 @@ int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (createReq.dnodeId != pDnode->dnodeId) { if (createReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to create bnode since %s, input:%d cur:%d", terrstr(), createReq.dnodeId, pDnode->dnodeId); dError("failed to create bnode since %s, input:%d cur:%d", terrstr(), createReq.dnodeId, pDnode->data.dnodeId);
return -1; return -1;
} else { } else {
return dndOpenNode(pWrapper); return dndOpenNode(pWrapper);
@ -72,7 +72,7 @@ int32_t bmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (dropReq.dnodeId != pDnode->dnodeId) { if (dropReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop bnode since %s", terrstr()); dError("failed to drop bnode since %s", terrstr());
return -1; return -1;

4
source/dnode/mgmt/bm/bmInt.c
View File

@ -113,8 +113,8 @@ void bmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = bmOpen; mgmtFp.openFp = bmOpen;
mgmtFp.closeFp = bmClose; mgmtFp.closeFp = bmClose;
mgmtFp.createMsgFp = bmProcessCreateReq; mgmtFp.createFp = bmProcessCreateReq;
mgmtFp.dropMsgFp = bmProcessDropReq; mgmtFp.dropFp = bmProcessDropReq;
mgmtFp.requiredFp = bmRequire; mgmtFp.requiredFp = bmRequire;
bmInitMsgHandle(pWrapper); bmInitMsgHandle(pWrapper);

36
source/dnode/mgmt/dm/dmFile.c
View File

@ -16,11 +16,11 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dmInt.h" #include "dmInt.h"
static void dmPrintDnodes(SDnodeMgmt *pMgmt); static void dmPrintDnodes(SDnodeData *pMgmt);
static bool dmIsEpChanged(SDnodeMgmt *pMgmt, int32_t dnodeId, const char *ep); static bool dmIsEpChanged(SDnodeData *pMgmt, int32_t dnodeId, const char *ep);
static void dmResetDnodes(SDnodeMgmt *pMgmt, SArray *dnodeEps); static void dmResetDnodes(SDnodeData *pMgmt, SArray *dnodeEps);
int32_t dmReadFile(SDnodeMgmt *pMgmt) { int32_t dmReadFile(SDnodeData *pMgmt) {
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT; int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
int32_t len = 0; int32_t len = 0;
int32_t maxLen = 256 * 1024; int32_t maxLen = 256 * 1024;
@ -62,21 +62,21 @@ int32_t dmReadFile(SDnodeMgmt *pMgmt) {
dError("failed to read %s since dnodeId not found", file); dError("failed to read %s since dnodeId not found", file);
goto PRASE_DNODE_OVER; goto PRASE_DNODE_OVER;
} }
pDnode->dnodeId = dnodeId->valueint; pDnode->data.dnodeId = dnodeId->valueint;
cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId"); cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
if (!clusterId || clusterId->type != cJSON_String) { if (!clusterId || clusterId->type != cJSON_String) {
dError("failed to read %s since clusterId not found", file); dError("failed to read %s since clusterId not found", file);
goto PRASE_DNODE_OVER; goto PRASE_DNODE_OVER;
} }
pDnode->clusterId = atoll(clusterId->valuestring); pDnode->data.clusterId = atoll(clusterId->valuestring);
cJSON *dropped = cJSON_GetObjectItem(root, "dropped"); cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_Number) { if (!dropped || dropped->type != cJSON_Number) {
dError("failed to read %s since dropped not found", file); dError("failed to read %s since dropped not found", file);
goto PRASE_DNODE_OVER; goto PRASE_DNODE_OVER;
} }
pDnode->dropped = dropped->valueint; pDnode->data.dropped = dropped->valueint;
cJSON *dnodes = cJSON_GetObjectItem(root, "dnodes"); cJSON *dnodes = cJSON_GetObjectItem(root, "dnodes");
if (!dnodes || dnodes->type != cJSON_Array) { if (!dnodes || dnodes->type != cJSON_Array) {
@ -138,15 +138,15 @@ PRASE_DNODE_OVER:
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile); if (pFile != NULL) taosCloseFile(&pFile);
if (dmIsEpChanged(pMgmt, pDnode->dnodeId, pDnode->localEp)) { if (dmIsEpChanged(pMgmt, pDnode->data.dnodeId, pDnode->data.localEp)) {
dError("localEp %s different with %s and need reconfigured", pDnode->localEp, file); dError("localEp %s different with %s and need reconfigured", pDnode->data.localEp, file);
return -1; return -1;
} }
if (taosArrayGetSize(pMgmt->dnodeEps) == 0) { if (taosArrayGetSize(pMgmt->dnodeEps) == 0) {
SDnodeEp dnodeEp = {0}; SDnodeEp dnodeEp = {0};
dnodeEp.isMnode = 1; dnodeEp.isMnode = 1;
taosGetFqdnPortFromEp(pDnode->firstEp, &dnodeEp.ep); taosGetFqdnPortFromEp(pDnode->data.firstEp, &dnodeEp.ep);
taosArrayPush(pMgmt->dnodeEps, &dnodeEp); taosArrayPush(pMgmt->dnodeEps, &dnodeEp);
} }
@ -156,7 +156,7 @@ PRASE_DNODE_OVER:
return code; return code;
} }
int32_t dmWriteFile(SDnodeMgmt *pMgmt) { int32_t dmWriteFile(SDnodeData *pMgmt) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
char file[PATH_MAX]; char file[PATH_MAX];
@ -174,9 +174,9 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt) {
char *content = taosMemoryCalloc(1, maxLen + 1); char *content = taosMemoryCalloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n"); len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", pDnode->dnodeId); len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", pDnode->data.dnodeId);
len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%" PRId64 "\",\n", pDnode->clusterId); len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%" PRId64 "\",\n", pDnode->data.clusterId);
len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pDnode->dropped); len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pDnode->data.dropped);
len += snprintf(content + len, maxLen - len, " \"dnodes\": [{\n"); len += snprintf(content + len, maxLen - len, " \"dnodes\": [{\n");
int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->dnodeEps); int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->dnodeEps);
@ -213,7 +213,7 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt) {
return 0; return 0;
} }
void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *dnodeEps) { void dmUpdateDnodeEps(SDnodeData *pMgmt, SArray *dnodeEps) {
int32_t numOfEps = taosArrayGetSize(dnodeEps); int32_t numOfEps = taosArrayGetSize(dnodeEps);
if (numOfEps <= 0) return; if (numOfEps <= 0) return;
@ -234,7 +234,7 @@ void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *dnodeEps) {
taosWUnLockLatch(&pMgmt->latch); taosWUnLockLatch(&pMgmt->latch);
} }
static void dmResetDnodes(SDnodeMgmt *pMgmt, SArray *dnodeEps) { static void dmResetDnodes(SDnodeData *pMgmt, SArray *dnodeEps) {
if (pMgmt->dnodeEps != dnodeEps) { if (pMgmt->dnodeEps != dnodeEps) {
SArray *tmp = pMgmt->dnodeEps; SArray *tmp = pMgmt->dnodeEps;
pMgmt->dnodeEps = taosArrayDup(dnodeEps); pMgmt->dnodeEps = taosArrayDup(dnodeEps);
@ -265,7 +265,7 @@ static void dmResetDnodes(SDnodeMgmt *pMgmt, SArray *dnodeEps) {
dmPrintDnodes(pMgmt); dmPrintDnodes(pMgmt);
} }
static void dmPrintDnodes(SDnodeMgmt *pMgmt) { static void dmPrintDnodes(SDnodeData *pMgmt) {
int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->dnodeEps); int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->dnodeEps);
dDebug("print dnode ep list, num:%d", numOfEps); dDebug("print dnode ep list, num:%d", numOfEps);
for (int32_t i = 0; i < numOfEps; i++) { for (int32_t i = 0; i < numOfEps; i++) {
@ -274,7 +274,7 @@ static void dmPrintDnodes(SDnodeMgmt *pMgmt) {
} }
} }
static bool dmIsEpChanged(SDnodeMgmt *pMgmt, int32_t dnodeId, const char *ep) { static bool dmIsEpChanged(SDnodeData *pMgmt, int32_t dnodeId, const char *ep) {
bool changed = false; bool changed = false;
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);

48
source/dnode/mgmt/dm/dmHandle.c
View File

@ -16,20 +16,20 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dmInt.h" #include "dmInt.h"
void dmSendStatusReq(SDnodeMgmt *pMgmt) { void dmSendStatusReq(SDnodeData *pMgmt) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
SStatusReq req = {0}; SStatusReq req = {0};
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
req.sver = tsVersion; req.sver = tsVersion;
req.dver = pMgmt->dver; req.dnodeVer = pMgmt->dnodeVer;
req.dnodeId = pDnode->dnodeId; req.dnodeId = pDnode->data.dnodeId;
req.clusterId = pDnode->clusterId; req.clusterId = pDnode->data.clusterId;
req.rebootTime = pDnode->rebootTime; req.rebootTime = pDnode->data.rebootTime;
req.updateTime = pMgmt->updateTime; req.updateTime = pMgmt->updateTime;
req.numOfCores = tsNumOfCores; req.numOfCores = tsNumOfCores;
req.numOfSupportVnodes = pDnode->numOfSupportVnodes; req.numOfSupportVnodes = pDnode->data.supportVnodes;
tstrncpy(req.dnodeEp, pDnode->localEp, TSDB_EP_LEN); tstrncpy(req.dnodeEp, pDnode->data.localEp, TSDB_EP_LEN);
req.clusterCfg.statusInterval = tsStatusInterval; req.clusterCfg.statusInterval = tsStatusInterval;
req.clusterCfg.checkTime = 0; req.clusterCfg.checkTime = 0;
@ -40,7 +40,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN); memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODE);
if (pWrapper != NULL) { if (pWrapper != NULL) {
SMonVloadInfo info = {0}; SMonVloadInfo info = {0};
dmGetVnodeLoads(pWrapper, &info); dmGetVnodeLoads(pWrapper, &info);
@ -62,34 +62,34 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
tmsgSendReq(&pMgmt->msgCb, &epSet, &rpcMsg); tmsgSendReq(&pMgmt->msgCb, &epSet, &rpcMsg);
} }
static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) { static void dmUpdateDnodeCfg(SDnodeData *pMgmt, SDnodeCfg *pCfg) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
if (pDnode->dnodeId == 0) { if (pDnode->data.dnodeId == 0) {
dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId); dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
taosWLockLatch(&pMgmt->latch); taosWLockLatch(&pMgmt->latch);
pDnode->dnodeId = pCfg->dnodeId; pDnode->data.dnodeId = pCfg->dnodeId;
pDnode->clusterId = pCfg->clusterId; pDnode->data.clusterId = pCfg->clusterId;
dmWriteFile(pMgmt); dmWriteFile(pMgmt);
taosWUnLockLatch(&pMgmt->latch); taosWUnLockLatch(&pMgmt->latch);
} }
} }
int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessStatusRsp(SDnodeData *pMgmt, SNodeMsg *pMsg) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
SRpcMsg *pRsp = &pMsg->rpcMsg; SRpcMsg *pRsp = &pMsg->rpcMsg;
if (pRsp->code != TSDB_CODE_SUCCESS) { if (pRsp->code != TSDB_CODE_SUCCESS) {
if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pDnode->dropped && pDnode->dnodeId > 0) { if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pDnode->data.dropped && pDnode->data.dnodeId > 0) {
dInfo("dnode:%d, set to dropped since not exist in mnode", pDnode->dnodeId); dInfo("dnode:%d, set to dropped since not exist in mnode", pDnode->data.dnodeId);
pDnode->dropped = 1; pDnode->data.dropped = 1;
dmWriteFile(pMgmt); dmWriteFile(pMgmt);
} }
} else { } else {
SStatusRsp statusRsp = {0}; SStatusRsp statusRsp = {0};
if (pRsp->pCont != NULL && pRsp->contLen != 0 && if (pRsp->pCont != NULL && pRsp->contLen != 0 &&
tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) { tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
pMgmt->dver = statusRsp.dver; pMgmt->dnodeVer = statusRsp.dnodeVer;
dmUpdateDnodeCfg(pMgmt, &statusRsp.dnodeCfg); dmUpdateDnodeCfg(pMgmt, &statusRsp.dnodeCfg);
dmUpdateDnodeEps(pMgmt, statusRsp.pDnodeEps); dmUpdateDnodeEps(pMgmt, statusRsp.pDnodeEps);
} }
@ -100,26 +100,26 @@ int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessAuthRsp(SDnodeData *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg; SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("auth rsp is received, but not supported yet"); dError("auth rsp is received, but not supported yet");
return 0; return 0;
} }
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessGrantRsp(SDnodeData *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg; SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("grant rsp is received, but not supported yet"); dError("grant rsp is received, but not supported yet");
return 0; return 0;
} }
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) { int32_t dmProcessConfigReq(SDnodeData *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pReq = &pMsg->rpcMsg; SRpcMsg *pReq = &pMsg->rpcMsg;
SDCfgDnodeReq *pCfg = pReq->pCont; SDCfgDnodeReq *pCfg = pReq->pCont;
dError("config req is received, but not supported yet"); dError("config req is received, but not supported yet");
return TSDB_CODE_OPS_NOT_SUPPORT; return TSDB_CODE_OPS_NOT_SUPPORT;
} }
static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *pMsg) { static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) { if (pWrapper != NULL) {
dndReleaseWrapper(pWrapper); dndReleaseWrapper(pWrapper);
@ -136,7 +136,7 @@ static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *
return -1; return -1;
} }
int32_t code = (*pWrapper->fp.createMsgFp)(pWrapper, pMsg); int32_t code = (*pWrapper->fp.createFp)(pWrapper, pMsg);
if (code != 0) { if (code != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr()); dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
} else { } else {
@ -147,7 +147,7 @@ static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *
return code; return code;
} }
static int32_t dmProcessDropNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *pMsg) { static int32_t dmProcessDropNodeMsg(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype); SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) { if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED; terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
@ -158,7 +158,7 @@ static int32_t dmProcessDropNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *pM
taosWLockLatch(&pWrapper->latch); taosWLockLatch(&pWrapper->latch);
pWrapper->deployed = false; pWrapper->deployed = false;
int32_t code = (*pWrapper->fp.dropMsgFp)(pWrapper, pMsg); int32_t code = (*pWrapper->fp.dropFp)(pWrapper, pMsg);
if (code != 0) { if (code != 0) {
pWrapper->deployed = true; pWrapper->deployed = true;
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr()); dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());

23
source/dnode/mgmt/dm/dmInt.c
View File

@ -16,13 +16,13 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dmInt.h" #include "dmInt.h"
void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet) { void dmGetMnodeEpSet(SDnodeData *pMgmt, SEpSet *pEpSet) {
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
*pEpSet = pMgmt->mnodeEpSet; *pEpSet = pMgmt->mnodeEpSet;
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
} }
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet) { void dmUpdateMnodeEpSet(SDnodeData *pMgmt, SEpSet *pEpSet) {
dInfo("mnode is changed, num:%d use:%d", pEpSet->numOfEps, pEpSet->inUse); dInfo("mnode is changed, num:%d use:%d", pEpSet->numOfEps, pEpSet->inUse);
taosWLockLatch(&pMgmt->latch); taosWLockLatch(&pMgmt->latch);
@ -35,7 +35,7 @@ void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet) {
} }
void dmGetDnodeEp(SMgmtWrapper *pWrapper, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) { void dmGetDnodeEp(SMgmtWrapper *pWrapper, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeData *pMgmt = pWrapper->pMgmt;
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t)); SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t));
@ -54,7 +54,7 @@ void dmGetDnodeEp(SMgmtWrapper *pWrapper, int32_t dnodeId, char *pEp, char *pFqd
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
} }
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pReq) { void dmSendRedirectRsp(SDnodeData *pMgmt, const SRpcMsg *pReq) {
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
SEpSet epSet = {0}; SEpSet epSet = {0};
@ -63,7 +63,7 @@ void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pReq) {
dDebug("RPC %p, req is redirected, num:%d use:%d", pReq->handle, epSet.numOfEps, epSet.inUse); dDebug("RPC %p, req is redirected, num:%d use:%d", pReq->handle, epSet.numOfEps, epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) { for (int32_t i = 0; i < epSet.numOfEps; ++i) {
dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port); dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
if (strcmp(epSet.eps[i].fqdn, pDnode->localFqdn) == 0 && epSet.eps[i].port == pDnode->serverPort) { if (strcmp(epSet.eps[i].fqdn, pDnode->data.localFqdn) == 0 && epSet.eps[i].port == pDnode->data.serverPort) {
epSet.inUse = (i + 1) % epSet.numOfEps; epSet.inUse = (i + 1) % epSet.numOfEps;
} }
@ -80,15 +80,14 @@ static int32_t dmStart(SMgmtWrapper *pWrapper) {
static int32_t dmInit(SMgmtWrapper *pWrapper) { static int32_t dmInit(SMgmtWrapper *pWrapper) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
SDnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SDnodeMgmt)); SDnodeData *pMgmt = taosMemoryCalloc(1, sizeof(SDnodeData));
dInfo("dnode-mgmt start to init"); dInfo("dnode-mgmt start to init");
pDnode->dnodeId = 0; pDnode->data.dnodeId = 0;
pDnode->dropped = 0; pDnode->data.dropped = 0;
pDnode->clusterId = 0; pDnode->data.clusterId = 0;
pMgmt->path = pWrapper->path; pMgmt->path = pWrapper->path;
pMgmt->pDnode = pDnode; pMgmt->pDnode = pDnode;
pMgmt->pWrapper = pWrapper;
taosInitRWLatch(&pMgmt->latch); taosInitRWLatch(&pMgmt->latch);
pMgmt->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK); pMgmt->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
@ -103,7 +102,7 @@ static int32_t dmInit(SMgmtWrapper *pWrapper) {
return -1; return -1;
} }
if (pDnode->dropped) { if (pDnode->data.dropped) {
dError("dnode will not start since its already dropped"); dError("dnode will not start since its already dropped");
return -1; return -1;
} }
@ -125,7 +124,7 @@ static int32_t dmInit(SMgmtWrapper *pWrapper) {
} }
static void dmCleanup(SMgmtWrapper *pWrapper) { static void dmCleanup(SMgmtWrapper *pWrapper) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeData *pMgmt = pWrapper->pMgmt;
if (pMgmt == NULL) return; if (pMgmt == NULL) return;
dInfo("dnode-mgmt start to clean up"); dInfo("dnode-mgmt start to clean up");

8
source/dnode/mgmt/dm/dmMonitor.c
View File

@ -18,13 +18,13 @@
static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->protocol = 1; pInfo->protocol = 1;
pInfo->dnode_id = pDnode->dnodeId; pInfo->dnode_id = pDnode->data.dnodeId;
pInfo->cluster_id = pDnode->clusterId; pInfo->cluster_id = pDnode->data.clusterId;
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN); tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
} }
static void dmGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) { static void dmGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pDnode->rebootTime) / (86400000.0f); pInfo->uptime = (taosGetTimestampMs() - pDnode->data.rebootTime) / (86400000.0f);
pInfo->has_mnode = pDnode->wrappers[MNODE].required; pInfo->has_mnode = pDnode->wrappers[MNODE].required;
pInfo->has_qnode = pDnode->wrappers[QNODE].required; pInfo->has_qnode = pDnode->wrappers[QNODE].required;
pInfo->has_snode = pDnode->wrappers[SNODE].required; pInfo->has_snode = pDnode->wrappers[SNODE].required;
@ -79,7 +79,7 @@ void dmSendMonitorReport(SDnode *pDnode) {
} }
} }
pWrapper = &pDnode->wrappers[VNODES]; pWrapper = &pDnode->wrappers[VNODE];
if (getFromAPI) { if (getFromAPI) {
if (dndMarkWrapper(pWrapper) == 0) { if (dndMarkWrapper(pWrapper) == 0) {
vmGetMonitorInfo(pWrapper, &vmInfo); vmGetMonitorInfo(pWrapper, &vmInfo);

16
source/dnode/mgmt/dm/dmWorker.c
View File

@ -17,7 +17,7 @@
#include "dmInt.h" #include "dmInt.h"
static void *dmThreadRoutine(void *param) { static void *dmThreadRoutine(void *param) {
SDnodeMgmt *pMgmt = param; SDnodeData *pMgmt = param;
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
int64_t lastStatusTime = taosGetTimestampMs(); int64_t lastStatusTime = taosGetTimestampMs();
int64_t lastMonitorTime = lastStatusTime; int64_t lastMonitorTime = lastStatusTime;
@ -27,7 +27,7 @@ static void *dmThreadRoutine(void *param) {
while (true) { while (true) {
taosThreadTestCancel(); taosThreadTestCancel();
taosMsleep(200); taosMsleep(200);
if (dndGetStatus(pDnode) != DND_STAT_RUNNING || pDnode->dropped) { if (dndGetStatus(pDnode) != DND_STAT_RUNNING || pDnode->data.dropped) {
continue; continue;
} }
@ -47,7 +47,7 @@ static void *dmThreadRoutine(void *param) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t dmStartThread(SDnodeMgmt *pMgmt) { int32_t dmStartThread(SDnodeData *pMgmt) {
pMgmt->threadId = taosCreateThread(dmThreadRoutine, pMgmt); pMgmt->threadId = taosCreateThread(dmThreadRoutine, pMgmt);
if (pMgmt->threadId == NULL) { if (pMgmt->threadId == NULL) {
dError("failed to init dnode thread"); dError("failed to init dnode thread");
@ -59,7 +59,7 @@ int32_t dmStartThread(SDnodeMgmt *pMgmt) {
} }
static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pInfo->ahandle; SDnodeData *pMgmt = pInfo->ahandle;
SDnode *pDnode = pMgmt->pDnode; SDnode *pDnode = pMgmt->pDnode;
SRpcMsg *pRpc = &pMsg->rpcMsg; SRpcMsg *pRpc = &pMsg->rpcMsg;
@ -95,7 +95,7 @@ static void dmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
int32_t dmStartWorker(SDnodeMgmt *pMgmt) { int32_t dmStartWorker(SDnodeData *pMgmt) {
SSingleWorkerCfg mcfg = {.min = 1, .max = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt}; SSingleWorkerCfg mcfg = {.min = 1, .max = 1, .name = "dnode-mgmt", .fp = (FItem)dmProcessQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mcfg) != 0) { if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mcfg) != 0) {
dError("failed to start dnode mgmt worker since %s", terrstr()); dError("failed to start dnode mgmt worker since %s", terrstr());
@ -112,7 +112,7 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
return 0; return 0;
} }
void dmStopWorker(SDnodeMgmt *pMgmt) { void dmStopWorker(SDnodeData *pMgmt) {
tSingleWorkerCleanup(&pMgmt->mgmtWorker); tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tSingleWorkerCleanup(&pMgmt->monitorWorker); tSingleWorkerCleanup(&pMgmt->monitorWorker);
@ -124,7 +124,7 @@ void dmStopWorker(SDnodeMgmt *pMgmt) {
} }
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeData *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
@ -133,7 +133,7 @@ int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt; SDnodeData *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker; SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name); dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);

10
source/dnode/mgmt/exe/dndMain.c
View File

@ -26,7 +26,7 @@ static struct {
char apolloUrl[PATH_MAX]; char apolloUrl[PATH_MAX];
SArray *pArgs; // SConfigPair SArray *pArgs; // SConfigPair
SDnode *pDnode; SDnode *pDnode;
EDndType ntype; EDndNodeType ntype;
} global = {0}; } global = {0};
static void dndStopDnode(int signum, void *info, void *ctx) { static void dndStopDnode(int signum, void *info, void *ctx) {
@ -46,7 +46,7 @@ static void dndSetSignalHandle() {
taosSetSignal(SIGQUIT, dndStopDnode); taosSetSignal(SIGQUIT, dndStopDnode);
if (!tsMultiProcess) { if (!tsMultiProcess) {
} else if (global.ntype == DNODE || global.ntype == NODE_MAX) { } else if (global.ntype == NODE_BEGIN || global.ntype == NODE_END) {
taosIgnSignal(SIGCHLD); taosIgnSignal(SIGCHLD);
} else { } else {
taosKillChildOnParentStopped(); taosKillChildOnParentStopped();
@ -72,8 +72,8 @@ static int32_t dndParseArgs(int32_t argc, char const *argv[]) {
tstrncpy(global.envFile, argv[++i], PATH_MAX); tstrncpy(global.envFile, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-n") == 0) { } else if (strcmp(argv[i], "-n") == 0) {
global.ntype = atoi(argv[++i]); global.ntype = atoi(argv[++i]);
if (global.ntype <= DNODE || global.ntype > NODE_MAX) { if (global.ntype <= NODE_BEGIN || global.ntype > NODE_END) {
printf("'-n' range is [1 - %d], default is 0\n", NODE_MAX - 1); printf("'-n' range is [1 - %d], default is 0\n", NODE_END - 1);
return -1; return -1;
} }
} else if (strcmp(argv[i], "-k") == 0) { } else if (strcmp(argv[i], "-k") == 0) {
@ -135,7 +135,7 @@ static int32_t dndInitLog() {
static void dndSetProcInfo(int32_t argc, char **argv) { static void dndSetProcInfo(int32_t argc, char **argv) {
taosSetProcPath(argc, argv); taosSetProcPath(argc, argv);
if (global.ntype != DNODE && global.ntype != NODE_MAX) { if (global.ntype != NODE_BEGIN && global.ntype != NODE_END) {
const char *name = dndNodeProcStr(global.ntype); const char *name = dndNodeProcStr(global.ntype);
taosSetProcName(argc, argv, name); taosSetProcName(argc, argv, name);
} }

39
source/dnode/mgmt/inc/dmInt.h
View File

@ -22,35 +22,20 @@
extern "C" { extern "C" {
#endif #endif
typedef struct SDnodeMgmt {
int64_t dver;
int64_t updateTime;
int8_t statusSent;
SEpSet mnodeEpSet;
SHashObj *dnodeHash;
SArray *dnodeEps;
TdThread *threadId;
SRWLatch latch;
SSingleWorker mgmtWorker;
SSingleWorker monitorWorker;
SMsgCb msgCb;
const char *path;
SDnode *pDnode;
SMgmtWrapper *pWrapper;
} SDnodeMgmt;
// dmFile.c // dmFile.c
int32_t dmReadFile(SDnodeMgmt *pMgmt); int32_t dmReadFile(SDnodeData *pMgmt);
int32_t dmWriteFile(SDnodeMgmt *pMgmt); int32_t dmWriteFile(SDnodeData *pMgmt);
void dmUpdateDnodeEps(SDnodeMgmt *pMgmt, SArray *pDnodeEps); void dmUpdateDnodeEps(SDnodeData *pMgmt, SArray *pDnodeEps);
// dmHandle.c // dmHandle.c
void dmInitMsgHandle(SMgmtWrapper *pWrapper); void dmInitMsgHandle(SMgmtWrapper *pWrapper);
void dmSendStatusReq(SDnodeMgmt *pMgmt); void dmSendStatusReq(SDnodeData *pMgmt);
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessConfigReq(SDnodeData *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessStatusRsp(SDnodeData *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessAuthRsp(SDnodeData *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg); int32_t dmProcessGrantRsp(SDnodeData *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessCDnodeReq(SDnode *pDnode, SNodeMsg *pMsg); int32_t dmProcessCDnodeReq(SDnode *pDnode, SNodeMsg *pMsg);
// dmMonitor.c // dmMonitor.c
@ -58,9 +43,9 @@ void dmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo);
void dmSendMonitorReport(SDnode *pDnode); void dmSendMonitorReport(SDnode *pDnode);
// dmWorker.c // dmWorker.c
int32_t dmStartThread(SDnodeMgmt *pMgmt); int32_t dmStartThread(SDnodeData *pMgmt);
int32_t dmStartWorker(SDnodeMgmt *pMgmt); int32_t dmStartWorker(SDnodeData *pMgmt);
void dmStopWorker(SDnodeMgmt *pMgmt); void dmStopWorker(SDnodeData *pMgmt);
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg); int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);

199
source/dnode/mgmt/inc/dndInt.h
View File

@ -1,199 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_INT_H_
#define _TD_DND_INT_H_
#include "os.h"
#include "cJSON.h"
#include "tcache.h"
#include "tcrc32c.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "thash.h"
#include "tlockfree.h"
#include "tlog.h"
#include "tmsg.h"
#include "tmsgcb.h"
#include "tprocess.h"
#include "tqueue.h"
#include "trpc.h"
#include "tthread.h"
#include "ttime.h"
#include "tworker.h"
#include "dnode.h"
#include "monitor.h"
#ifdef __cplusplus
extern "C" {
#endif
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
typedef enum { DNODE, VNODES, QNODE, SNODE, MNODE, BNODE, NODE_MAX } EDndType;
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndStatus;
typedef enum { DND_ENV_INIT, DND_ENV_READY, DND_ENV_CLEANUP } EEnvStatus;
typedef enum { PROC_SINGLE, PROC_CHILD, PROC_PARENT } EProcType;
typedef struct SMgmtFp SMgmtFp;
typedef struct SMgmtWrapper SMgmtWrapper;
typedef struct SMsgHandle SMsgHandle;
typedef struct SDnodeMgmt SDnodeMgmt;
typedef struct SVnodesMgmt SVnodesMgmt;
typedef struct SMnodeMgmt SMnodeMgmt;
typedef struct SQnodeMgmt SQnodeMgmt;
typedef struct SSnodeMgmt SSnodeMgmt;
typedef struct SBnodeMgmt SBnodeMgmt;
typedef int32_t (*NodeMsgFp)(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*OpenNodeFp)(SMgmtWrapper *pWrapper);
typedef void (*CloseNodeFp)(SMgmtWrapper *pWrapper);
typedef int32_t (*StartNodeFp)(SMgmtWrapper *pWrapper);
typedef int32_t (*CreateNodeFp)(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*DropNodeFp)(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*RequireNodeFp)(SMgmtWrapper *pWrapper, bool *required);
typedef struct SMsgHandle {
SMgmtWrapper *pQndWrapper;
SMgmtWrapper *pMndWrapper;
SMgmtWrapper *pWrapper;
} SMsgHandle;
typedef struct SMgmtFp {
OpenNodeFp openFp;
CloseNodeFp closeFp;
StartNodeFp startFp;
CreateNodeFp createMsgFp;
DropNodeFp dropMsgFp;
RequireNodeFp requiredFp;
} SMgmtFp;
typedef struct SMgmtWrapper {
const char *name;
char *path;
int32_t refCount;
SRWLatch latch;
EDndType ntype;
bool deployed;
bool required;
EProcType procType;
int32_t procId;
SProcObj *pProc;
SShm shm;
void *pMgmt;
SDnode *pDnode;
SMgmtFp fp;
int8_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
NodeMsgFp msgFps[TDMT_MAX];
} SMgmtWrapper;
typedef struct {
void *serverRpc;
void *clientRpc;
SMsgHandle msgHandles[TDMT_MAX];
} STransMgmt;
typedef struct SDnode {
int64_t clusterId;
int32_t dnodeId;
int32_t numOfSupportVnodes;
int64_t rebootTime;
char *localEp;
char *localFqdn;
char *firstEp;
char *secondEp;
char *dataDir;
SDiskCfg *disks;
int32_t numOfDisks;
uint16_t serverPort;
bool dropped;
EProcType procType;
EDndType ntype;
EDndStatus status;
EDndEvent event;
SStartupReq startup;
TdFilePtr lockfile;
STransMgmt trans;
SMgmtWrapper wrappers[NODE_MAX];
} SDnode;
// dndEnv.c
const char *dndStatStr(EDndStatus stat);
const char *dndNodeLogStr(EDndType ntype);
const char *dndNodeProcStr(EDndType ntype);
const char *dndEventStr(EDndEvent ev);
// dndExec.c
int32_t dndOpenNode(SMgmtWrapper *pWrapper);
void dndCloseNode(SMgmtWrapper *pWrapper);
// dndFile.c
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed);
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed);
TdFilePtr dndCheckRunning(const char *dataDir);
int32_t dndReadShmFile(SDnode *pDnode);
int32_t dndWriteShmFile(SDnode *pDnode);
// dndInt.c
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, EDndType nType);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper);
void dndReleaseWrapper(SMgmtWrapper *pWrapper);
void dndHandleEvent(SDnode *pDnode, EDndEvent event);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndTransport.c
int32_t dndInitTrans(SDnode *pDnode);
void dndCleanupTrans(SDnode *pDnode);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper);
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRecv(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
// mgmt
void dmSetMgmtFp(SMgmtWrapper *pWrapper);
void bmSetMgmtFp(SMgmtWrapper *pWrapper);
void qmSetMgmtFp(SMgmtWrapper *pMgmt);
void smSetMgmtFp(SMgmtWrapper *pWrapper);
void vmSetMgmtFp(SMgmtWrapper *pWrapper);
void mmSetMgmtFp(SMgmtWrapper *pMgmt);
void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pMsg);
void dmGetMonitorSysInfo(SMonSysInfo *pInfo);
void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo);
void mmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonMmInfo *mmInfo);
void vmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonVmInfo *vmInfo);
void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo);
void smGetMonitorInfo(SMgmtWrapper *pWrapper, SMonSmInfo *smInfo);
void bmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonBmInfo *bmInfo);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_INT_H_*/

10
source/dnode/mgmt/interface/CMakeLists.txt Normal file
View File

@ -0,0 +1,10 @@
aux_source_directory(src DNODE_INTERFACE)
add_library(dnode_interface STATIC ${DNODE_INTERFACE})
target_include_directories(
dnode_interface
PUBLIC "${TD_SOURCE_DIR}/include/dnode/mgmt"
PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
dnode_interface cjson mnode vnode qnode snode bnode wal sync taos tfs monitor util
)

153
source/dnode/mgmt/interface/inc/dndDef.h Normal file
View File

@ -0,0 +1,153 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_DEF_H_
#define _TD_DND_DEF_H_
#include "dndLog.h"
#include "cJSON.h"
#include "tcache.h"
#include "tcrc32c.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "thash.h"
#include "tlockfree.h"
#include "tlog.h"
#include "tmsg.h"
#include "tmsgcb.h"
#include "tprocess.h"
#include "tqueue.h"
#include "trpc.h"
#include "tthread.h"
#include "ttime.h"
#include "tworker.h"
#include "dnode.h"
#include "monitor.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef enum { NODE_BEGIN, VNODE, QNODE, SNODE, MNODE, BNODE, NODE_END } EDndNodeType;
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EDndRunStatus;
typedef enum { DND_ENV_INIT, DND_ENV_READY, DND_ENV_CLEANUP } EDndEnvStatus;
typedef enum { DND_PROC_SINGLE, DND_PROC_CHILD, DND_PROC_PARENT } EDndProcType;
typedef int32_t (*NodeMsgFp)(struct SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*OpenNodeFp)(struct SMgmtWrapper *pWrapper);
typedef void (*CloseNodeFp)(struct SMgmtWrapper *pWrapper);
typedef int32_t (*StartNodeFp)(struct SMgmtWrapper *pWrapper);
typedef void (*StopNodeFp)(struct SMgmtWrapper *pWrapper);
typedef int32_t (*CreateNodeFp)(struct SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*DropNodeFp)(struct SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
typedef int32_t (*RequireNodeFp)(struct SMgmtWrapper *pWrapper, bool *required);
typedef struct {
SMgmtWrapper *pQndWrapper;
SMgmtWrapper *pMndWrapper;
SMgmtWrapper *pNdWrapper;
} SMsgHandle;
typedef struct {
OpenNodeFp openFp;
CloseNodeFp closeFp;
StartNodeFp startFp;
StopNodeFp stopFp;
CreateNodeFp createFp;
DropNodeFp dropFp;
RequireNodeFp requiredFp;
} SMgmtFp;
typedef struct SMgmtWrapper {
SDnode *pDnode;
struct {
const char *name;
char *path;
int32_t refCount;
SRWLatch latch;
EDndNodeType ntype;
bool deployed;
bool required;
SMgmtFp fp;
void *pMgmt;
};
struct {
EDndProcType procType;
int32_t procId;
SProcObj *procObj;
SShm procShm;
};
struct {
int8_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
NodeMsgFp msgFps[TDMT_MAX];
};
} SMgmtWrapper;
typedef struct {
void *serverRpc;
void *clientRpc;
SMsgHandle msgHandles[TDMT_MAX];
} SDnodeTrans;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
int64_t dnodeVer;
int64_t updateTime;
int64_t rebootTime;
bool dropped;
int8_t statusSent;
SEpSet mnodeEpSet;
SHashObj *dnodeHash;
SArray *dnodeEps;
TdThread *threadId;
SRWLatch latch;
SSingleWorker mgmtWorker;
SSingleWorker monitorWorker;
SMsgCb msgCb;
SDnode *pDnode;
const char *path;
TdFilePtr lockfile;
struct {
char *localEp;
char *localFqdn;
char *firstEp;
char *secondEp;
char *dataDir;
SDiskCfg *disks;
int32_t numOfDisks;
int32_t supportVnodes;
uint16_t serverPort;
};
} SDnodeData;
typedef struct SDnode {
EDndProcType ptype;
EDndNodeType ntype;
EDndRunStatus status;
EDndEvent event;
SStartupReq startup;
SDnodeTrans trans;
SDnodeData data;
SMgmtWrapper wrappers[NODE_END];
} SDnode;
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_DEF_H_*/

86
source/dnode/mgmt/interface/inc/dndInt.h Normal file
View File

@ -0,0 +1,86 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_INT_H_
#define _TD_DND_INT_H_
#include "dndLog.h"
#include "dndDef.h"
#ifdef __cplusplus
extern "C" {
#endif
// dndEnv.c
const char *dndStatStr(EDndRunStatus stat);
const char *dndNodeLogStr(EDndNodeType ntype);
const char *dndNodeProcStr(EDndNodeType ntype);
const char *dndEventStr(EDndEvent ev);
// dndExec.c
int32_t dndOpenNode(SMgmtWrapper *pWrapper);
void dndCloseNode(SMgmtWrapper *pWrapper);
// dndFile.c
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed);
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed);
TdFilePtr dndCheckRunning(const char *dataDir);
int32_t dndReadShmFile(SDnode *pDnode);
int32_t dndWriteShmFile(SDnode *pDnode);
// dndInt.c
EDndRunStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndRunStatus stat);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId);
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, EDndNodeType nType);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper);
void dndReleaseWrapper(SMgmtWrapper *pWrapper);
void dndHandleEvent(SDnode *pDnode, EDndEvent event);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndTransport.c
int32_t dndInitTrans(SDnode *pDnode);
void dndCleanupTrans(SDnode *pDnode);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper);
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRecv(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
// mgmt
void dmSetMgmtFp(SMgmtWrapper *pWrapper);
void bmSetMgmtFp(SMgmtWrapper *pWrapper);
void qmSetMgmtFp(SMgmtWrapper *pMgmt);
void smSetMgmtFp(SMgmtWrapper *pWrapper);
void vmSetMgmtFp(SMgmtWrapper *pWrapper);
void mmSetMgmtFp(SMgmtWrapper *pMgmt);
void dmGetMnodeEpSet(SDnodeData *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeData *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeData *pMgmt, const SRpcMsg *pMsg);
void dmGetMonitorSysInfo(SMonSysInfo *pInfo);
void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo);
void mmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonMmInfo *mmInfo);
void vmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonVmInfo *vmInfo);
void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo);
void smGetMonitorInfo(SMgmtWrapper *pWrapper, SMonSmInfo *smInfo);
void bmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonBmInfo *bmInfo);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_INT_H_*/

36
source/dnode/mgmt/interface/inc/dndLog.h Normal file
View File

@ -0,0 +1,36 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_LOG_H_
#define _TD_DND_LOG_H_
#include "tlog.h"
#ifdef __cplusplus
extern "C" {
#endif
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_LOG_H_*/

86
source/dnode/mgmt/interface/src/dndInt.c Normal file
View File

@ -0,0 +1,86 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
EDndRunStatus dndGetStatus(SDnode *pDnode) { return pDnode->status; }
void dndSetStatus(SDnode *pDnode, EDndRunStatus status) {
if (pDnode->status != status) {
dDebug("dnode status set from %s to %s", dndStatStr(pDnode->status), dndStatStr(status));
pDnode->status = status;
}
}
const char *dndStatStr(EDndRunStatus status) {
switch (status) {
case DND_STAT_INIT:
return "init";
case DND_STAT_RUNNING:
return "running";
case DND_STAT_STOPPED:
return "stopped";
default:
return "UNKNOWN";
}
}
const char *dndNodeLogStr(EDndNodeType ntype) {
switch (ntype) {
case VNODE:
return "vnode";
case QNODE:
return "qnode";
case SNODE:
return "snode";
case MNODE:
return "mnode";
case BNODE:
return "bnode";
default:
return "taosd";
}
}
const char *dndNodeProcStr(EDndNodeType ntype) {
switch (ntype) {
case VNODE:
return "taosv";
case QNODE:
return "taosq";
case SNODE:
return "taoss";
case MNODE:
return "taosm";
case BNODE:
return "taosb";
default:
return "taosd";
}
}
const char *dndEventStr(EDndEvent ev) {
switch (ev) {
case DND_EVENT_START:
return "start";
case DND_EVENT_STOP:
return "stop";
case DND_EVENT_CHILD:
return "child";
default:
return "UNKNOWN";
}
}

60
source/dnode/mgmt/main/dndEnv.c
View File

@ -57,63 +57,3 @@ void dndCleanup() {
taosStopCacheRefreshWorker(); taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up"); dInfo("dnode env is cleaned up");
} }
const char *dndStatStr(EDndStatus status) {
switch (status) {
case DND_STAT_INIT:
return "init";
case DND_STAT_RUNNING:
return "running";
case DND_STAT_STOPPED:
return "stopped";
default:
return "UNKNOWN";
}
}
const char *dndNodeLogStr(EDndType ntype) {
switch (ntype) {
case VNODES:
return "vnode";
case QNODE:
return "qnode";
case SNODE:
return "snode";
case MNODE:
return "mnode";
case BNODE:
return "bnode";
default:
return "taosd";
}
}
const char *dndNodeProcStr(EDndType ntype) {
switch (ntype) {
case VNODES:
return "taosv";
case QNODE:
return "taosq";
case SNODE:
return "taoss";
case MNODE:
return "taosm";
case BNODE:
return "taosb";
default:
return "taosd";
}
}
const char *dndEventStr(EDndEvent ev) {
switch (ev) {
case DND_EVENT_START:
return "start";
case DND_EVENT_STOP:
return "stop";
case DND_EVENT_CHILD:
return "child";
default:
return "UNKNOWN";
}
}

66
source/dnode/mgmt/main/dndExec.c
View File

@ -29,7 +29,7 @@ static bool dndRequireNode(SMgmtWrapper *pWrapper) {
static int32_t dndInitNodeProc(SMgmtWrapper *pWrapper) { static int32_t dndInitNodeProc(SMgmtWrapper *pWrapper) {
int32_t shmsize = tsMnodeShmSize; int32_t shmsize = tsMnodeShmSize;
if (pWrapper->ntype == VNODES) { if (pWrapper->ntype == VNODE) {
shmsize = tsVnodeShmSize; shmsize = tsVnodeShmSize;
} else if (pWrapper->ntype == QNODE) { } else if (pWrapper->ntype == QNODE) {
shmsize = tsQnodeShmSize; shmsize = tsQnodeShmSize;
@ -43,18 +43,18 @@ static int32_t dndInitNodeProc(SMgmtWrapper *pWrapper) {
return -1; return -1;
} }
if (taosCreateShm(&pWrapper->shm, pWrapper->ntype, shmsize) != 0) { if (taosCreateShm(&pWrapper->procShm, pWrapper->ntype, shmsize) != 0) {
terrno = TAOS_SYSTEM_ERROR(terrno); terrno = TAOS_SYSTEM_ERROR(terrno);
dError("node:%s, failed to create shm size:%d since %s", pWrapper->name, shmsize, terrstr()); dError("node:%s, failed to create shm size:%d since %s", pWrapper->name, shmsize, terrstr());
return -1; return -1;
} }
dInfo("node:%s, shm:%d is created, size:%d", pWrapper->name, pWrapper->shm.id, shmsize); dInfo("node:%s, shm:%d is created, size:%d", pWrapper->name, pWrapper->procShm.id, shmsize);
SProcCfg cfg = dndGenProcCfg(pWrapper); SProcCfg cfg = dndGenProcCfg(pWrapper);
cfg.isChild = false; cfg.isChild = false;
pWrapper->procType = PROC_PARENT; pWrapper->procType = DND_PROC_PARENT;
pWrapper->pProc = taosProcInit(&cfg); pWrapper->procObj = taosProcInit(&cfg);
if (pWrapper->pProc == NULL) { if (pWrapper->procObj == NULL) {
dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr()); dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr());
return -1; return -1;
} }
@ -62,7 +62,7 @@ static int32_t dndInitNodeProc(SMgmtWrapper *pWrapper) {
return 0; return 0;
} }
static int32_t dndNewNodeProc(SMgmtWrapper *pWrapper, EDndType n) { static int32_t dndNewNodeProc(SMgmtWrapper *pWrapper, EDndNodeType n) {
char tstr[8] = {0}; char tstr[8] = {0};
char *args[6] = {0}; char *args[6] = {0};
snprintf(tstr, sizeof(tstr), "%d", n); snprintf(tstr, sizeof(tstr), "%d", n);
@ -85,7 +85,7 @@ static int32_t dndNewNodeProc(SMgmtWrapper *pWrapper, EDndType n) {
} }
static int32_t dndRunNodeProc(SMgmtWrapper *pWrapper) { static int32_t dndRunNodeProc(SMgmtWrapper *pWrapper) {
if (pWrapper->pDnode->ntype == NODE_MAX) { if (pWrapper->pDnode->ntype == NODE_END) {
dInfo("node:%s, should be started manually", pWrapper->name); dInfo("node:%s, should be started manually", pWrapper->name);
} else { } else {
if (dndNewNodeProc(pWrapper, pWrapper->ntype) != 0) { if (dndNewNodeProc(pWrapper, pWrapper->ntype) != 0) {
@ -93,7 +93,7 @@ static int32_t dndRunNodeProc(SMgmtWrapper *pWrapper) {
} }
} }
if (taosProcRun(pWrapper->pProc) != 0) { if (taosProcRun(pWrapper->procObj) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr()); dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1; return -1;
} }
@ -120,9 +120,9 @@ static int32_t dndOpenNodeImp(SMgmtWrapper *pWrapper) {
int32_t dndOpenNode(SMgmtWrapper *pWrapper) { int32_t dndOpenNode(SMgmtWrapper *pWrapper) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
if (pDnode->procType == PROC_SINGLE) { if (pDnode->ptype == DND_PROC_SINGLE) {
return dndOpenNodeImp(pWrapper); return dndOpenNodeImp(pWrapper);
} else if (pDnode->procType == PROC_PARENT) { } else if (pDnode->ptype == DND_PROC_PARENT) {
if (dndInitNodeProc(pWrapper) != 0) return -1; if (dndInitNodeProc(pWrapper) != 0) return -1;
if (dndWriteShmFile(pDnode) != 0) return -1; if (dndWriteShmFile(pDnode) != 0) return -1;
if (dndRunNodeProc(pWrapper) != 0) return -1; if (dndRunNodeProc(pWrapper) != 0) return -1;
@ -144,15 +144,15 @@ static void dndCloseNodeImp(SMgmtWrapper *pWrapper) {
taosMsleep(10); taosMsleep(10);
} }
if (pWrapper->pProc) { if (pWrapper->procObj) {
taosProcCleanup(pWrapper->pProc); taosProcCleanup(pWrapper->procObj);
pWrapper->pProc = NULL; pWrapper->procObj = NULL;
} }
dDebug("node:%s, mgmt has been closed", pWrapper->name); dDebug("node:%s, mgmt has been closed", pWrapper->name);
} }
void dndCloseNode(SMgmtWrapper *pWrapper) { void dndCloseNode(SMgmtWrapper *pWrapper) {
if (pWrapper->pDnode->procType == PROC_PARENT) { if (pWrapper->pDnode->ptype == DND_PROC_PARENT) {
if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) { if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) {
dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId); dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId);
taosKillProc(pWrapper->procId); taosKillProc(pWrapper->procId);
@ -172,9 +172,9 @@ static void dndProcessProcHandle(void *handle) {
static int32_t dndRunInSingleProcess(SDnode *pDnode) { static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dInfo("dnode run in single process"); dInfo("dnode run in single process");
pDnode->procType = PROC_SINGLE; pDnode->ptype = DND_PROC_SINGLE;
for (EDndType n = DNODE; n < NODE_MAX; ++n) { for (EDndNodeType n = NODE_BEGIN; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = dndRequireNode(pWrapper); pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
@ -187,7 +187,7 @@ static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dndSetStatus(pDnode, DND_STAT_RUNNING); dndSetStatus(pDnode, DND_STAT_RUNNING);
for (EDndType n = 0; n < NODE_MAX; ++n) { for (EDndNodeType n = 0; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
if (pWrapper->fp.startFp == NULL) continue; if (pWrapper->fp.startFp == NULL) continue;
@ -213,15 +213,15 @@ static int32_t dndRunInSingleProcess(SDnode *pDnode) {
static int32_t dndRunInParentProcess(SDnode *pDnode) { static int32_t dndRunInParentProcess(SDnode *pDnode) {
dInfo("dnode run in parent process"); dInfo("dnode run in parent process");
pDnode->procType = PROC_PARENT; pDnode->ptype = DND_PROC_PARENT;
SMgmtWrapper *pDWrapper = &pDnode->wrappers[DNODE]; SMgmtWrapper *pDWrapper = &pDnode->wrappers[NODE_BEGIN];
if (dndOpenNodeImp(pDWrapper) != 0) { if (dndOpenNodeImp(pDWrapper) != 0) {
dError("node:%s, failed to start since %s", pDWrapper->name, terrstr()); dError("node:%s, failed to start since %s", pDWrapper->name, terrstr());
return -1; return -1;
} }
for (EDndType n = DNODE + 1; n < NODE_MAX; ++n) { for (EDndNodeType n = NODE_BEGIN + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = dndRequireNode(pWrapper); pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
@ -233,7 +233,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
return -1; return -1;
} }
for (EDndType n = DNODE + 1; n < NODE_MAX; ++n) { for (EDndNodeType n = NODE_BEGIN + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
if (dndRunNodeProc(pWrapper) != 0) return -1; if (dndRunNodeProc(pWrapper) != 0) return -1;
@ -254,10 +254,10 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
dInfo("dnode is about to stop"); dInfo("dnode is about to stop");
dndSetStatus(pDnode, DND_STAT_STOPPED); dndSetStatus(pDnode, DND_STAT_STOPPED);
for (EDndType n = DNODE + 1; n < NODE_MAX; ++n) { for (EDndNodeType n = NODE_BEGIN + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
if (pDnode->ntype == NODE_MAX) continue; if (pDnode->ntype == NODE_END) continue;
if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) { if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) {
dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId); dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId);
@ -269,14 +269,14 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
} }
break; break;
} else { } else {
for (EDndType n = DNODE + 1; n < NODE_MAX; ++n) { for (EDndNodeType n = NODE_BEGIN + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue; if (!pWrapper->required) continue;
if (pDnode->ntype == NODE_MAX) continue; if (pDnode->ntype == NODE_END) continue;
if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) { if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId); dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
taosProcCloseHandles(pWrapper->pProc, dndProcessProcHandle); taosProcCloseHandles(pWrapper->procObj, dndProcessProcHandle);
dndNewNodeProc(pWrapper, n); dndNewNodeProc(pWrapper, n);
} }
} }
@ -291,7 +291,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
static int32_t dndRunInChildProcess(SDnode *pDnode) { static int32_t dndRunInChildProcess(SDnode *pDnode) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
dInfo("%s run in child process", pWrapper->name); dInfo("%s run in child process", pWrapper->name);
pDnode->procType = PROC_CHILD; pDnode->ptype = DND_PROC_CHILD;
pWrapper->required = dndRequireNode(pWrapper); pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) { if (!pWrapper->required) {
@ -301,7 +301,7 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) {
SMsgCb msgCb = dndCreateMsgcb(pWrapper); SMsgCb msgCb = dndCreateMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&msgCb); tmsgSetDefaultMsgCb(&msgCb);
pWrapper->procType = PROC_CHILD; pWrapper->procType = DND_PROC_CHILD;
if (dndOpenNodeImp(pWrapper) != 0) { if (dndOpenNodeImp(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr()); dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
@ -310,8 +310,8 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) {
SProcCfg cfg = dndGenProcCfg(pWrapper); SProcCfg cfg = dndGenProcCfg(pWrapper);
cfg.isChild = true; cfg.isChild = true;
pWrapper->pProc = taosProcInit(&cfg); pWrapper->procObj = taosProcInit(&cfg);
if (pWrapper->pProc == NULL) { if (pWrapper->procObj == NULL) {
dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr()); dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr());
return -1; return -1;
} }
@ -325,7 +325,7 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) {
dndSetStatus(pDnode, DND_STAT_RUNNING); dndSetStatus(pDnode, DND_STAT_RUNNING);
if (taosProcRun(pWrapper->pProc) != 0) { if (taosProcRun(pWrapper->procObj) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr()); dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1; return -1;
} }
@ -347,7 +347,7 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) {
int32_t dndRun(SDnode *pDnode) { int32_t dndRun(SDnode *pDnode) {
if (!tsMultiProcess) { if (!tsMultiProcess) {
return dndRunInSingleProcess(pDnode); return dndRunInSingleProcess(pDnode);
} else if (pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) { } else if (pDnode->ntype == NODE_BEGIN || pDnode->ntype == NODE_END) {
return dndRunInParentProcess(pDnode); return dndRunInParentProcess(pDnode);
} else { } else {
return dndRunInChildProcess(pDnode); return dndRunInChildProcess(pDnode);

38
source/dnode/mgmt/main/dndFile.c
View File

@ -148,7 +148,7 @@ int32_t dndReadShmFile(SDnode *pDnode) {
cJSON *root = NULL; cJSON *root = NULL;
TdFilePtr pFile = NULL; TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP); snprintf(file, sizeof(file), "%s%s.shmfile", pDnode->data.dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_READ); pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) { if (pFile == NULL) {
dDebug("file %s not exist", file); dDebug("file %s not exist", file);
@ -164,37 +164,37 @@ int32_t dndReadShmFile(SDnode *pDnode) {
goto _OVER; goto _OVER;
} }
for (EDndType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) { for (EDndNodeType ntype = NODE_BEGIN + 1; ntype < NODE_END; ++ntype) {
snprintf(itemName, sizeof(itemName), "%s_shmid", dndNodeProcStr(ntype)); snprintf(itemName, sizeof(itemName), "%s_shmid", dndNodeProcStr(ntype));
cJSON *shmid = cJSON_GetObjectItem(root, itemName); cJSON *shmid = cJSON_GetObjectItem(root, itemName);
if (shmid && shmid->type == cJSON_Number) { if (shmid && shmid->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.id = shmid->valueint; pDnode->wrappers[ntype].procShm.id = shmid->valueint;
} }
snprintf(itemName, sizeof(itemName), "%s_shmsize", dndNodeProcStr(ntype)); snprintf(itemName, sizeof(itemName), "%s_shmsize", dndNodeProcStr(ntype));
cJSON *shmsize = cJSON_GetObjectItem(root, itemName); cJSON *shmsize = cJSON_GetObjectItem(root, itemName);
if (shmsize && shmsize->type == cJSON_Number) { if (shmsize && shmsize->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.size = shmsize->valueint; pDnode->wrappers[ntype].procShm.size = shmsize->valueint;
} }
} }
} }
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) { if (!tsMultiProcess || pDnode->ntype == NODE_BEGIN || pDnode->ntype == NODE_END) {
for (EDndType ntype = DNODE; ntype < NODE_MAX; ++ntype) { for (EDndNodeType ntype = NODE_BEGIN; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
if (pWrapper->shm.id >= 0) { if (pWrapper->procShm.id >= 0) {
dDebug("shmid:%d, is closed, size:%d", pWrapper->shm.id, pWrapper->shm.size); dDebug("shmid:%d, is closed, size:%d", pWrapper->procShm.id, pWrapper->procShm.size);
taosDropShm(&pWrapper->shm); taosDropShm(&pWrapper->procShm);
} }
} }
} else { } else {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (taosAttachShm(&pWrapper->shm) != 0) { if (taosAttachShm(&pWrapper->procShm) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
dError("shmid:%d, failed to attach shm since %s", pWrapper->shm.id, terrstr()); dError("shmid:%d, failed to attach shm since %s", pWrapper->procShm.id, terrstr());
goto _OVER; goto _OVER;
} }
dInfo("node:%s, shmid:%d is attached, size:%d", pWrapper->name, pWrapper->shm.id, pWrapper->shm.size); dInfo("node:%s, shmid:%d is attached, size:%d", pWrapper->name, pWrapper->procShm.id, pWrapper->procShm.size);
} }
dDebug("successed to load %s", file); dDebug("successed to load %s", file);
@ -215,8 +215,8 @@ int32_t dndWriteShmFile(SDnode *pDnode) {
char realfile[PATH_MAX] = {0}; char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL; TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.shmfile.bak", pDnode->dataDir, TD_DIRSEP); snprintf(file, sizeof(file), "%s%s.shmfile.bak", pDnode->data.dataDir, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP); snprintf(realfile, sizeof(realfile), "%s%s.shmfile", pDnode->data.dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) { if (pFile == NULL) {
@ -226,13 +226,13 @@ int32_t dndWriteShmFile(SDnode *pDnode) {
} }
len += snprintf(content + len, MAXLEN - len, "{\n"); len += snprintf(content + len, MAXLEN - len, "{\n");
for (EDndType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) { for (EDndNodeType ntype = NODE_BEGIN + 1; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\":%d,\n", dndNodeProcStr(ntype), pWrapper->shm.id); len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\":%d,\n", dndNodeProcStr(ntype), pWrapper->procShm.id);
if (ntype == NODE_MAX - 1) { if (ntype == NODE_END - 1) {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d\n", dndNodeProcStr(ntype), pWrapper->shm.size); len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d\n", dndNodeProcStr(ntype), pWrapper->procShm.size);
} else { } else {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d,\n", dndNodeProcStr(ntype), pWrapper->shm.size); len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d,\n", dndNodeProcStr(ntype), pWrapper->procShm.size);
} }
} }
len += snprintf(content + len, MAXLEN - len, "}\n"); len += snprintf(content + len, MAXLEN - len, "}\n");

77
source/dnode/mgmt/main/dndInt.c → source/dnode/mgmt/main/dndObj.c
View File

@ -17,27 +17,27 @@
#include "dndInt.h" #include "dndInt.h"
static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) { static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes; pDnode->data.supportVnodes = pOption->numOfSupportVnodes;
pDnode->serverPort = pOption->serverPort; pDnode->data.serverPort = pOption->serverPort;
pDnode->dataDir = strdup(pOption->dataDir); pDnode->data.dataDir = strdup(pOption->dataDir);
pDnode->localEp = strdup(pOption->localEp); pDnode->data.localEp = strdup(pOption->localEp);
pDnode->localFqdn = strdup(pOption->localFqdn); pDnode->data.localFqdn = strdup(pOption->localFqdn);
pDnode->firstEp = strdup(pOption->firstEp); pDnode->data.firstEp = strdup(pOption->firstEp);
pDnode->secondEp = strdup(pOption->secondEp); pDnode->data.secondEp = strdup(pOption->secondEp);
pDnode->disks = pOption->disks; pDnode->data.disks = pOption->disks;
pDnode->numOfDisks = pOption->numOfDisks; pDnode->data.numOfDisks = pOption->numOfDisks;
pDnode->ntype = pOption->ntype; pDnode->ntype = pOption->ntype;
pDnode->rebootTime = taosGetTimestampMs(); pDnode->data.rebootTime = taosGetTimestampMs();
if (pDnode->dataDir == NULL || pDnode->localEp == NULL || pDnode->localFqdn == NULL || pDnode->firstEp == NULL || if (pDnode->data.dataDir == NULL || pDnode->data.localEp == NULL || pDnode->data.localFqdn == NULL || pDnode->data.firstEp == NULL ||
pDnode->secondEp == NULL) { pDnode->data.secondEp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) { if (!tsMultiProcess || pDnode->ntype == NODE_BEGIN || pDnode->ntype == NODE_END) {
pDnode->lockfile = dndCheckRunning(pDnode->dataDir); pDnode->data.lockfile = dndCheckRunning(pDnode->data.dataDir);
if (pDnode->lockfile == NULL) { if (pDnode->data.lockfile == NULL) {
return -1; return -1;
} }
} }
@ -46,20 +46,20 @@ static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
} }
static void dndClearVars(SDnode *pDnode) { static void dndClearVars(SDnode *pDnode) {
for (EDndType n = 0; n < NODE_MAX; ++n) { for (EDndNodeType n = 0; n < NODE_END; ++n) {
SMgmtWrapper *pMgmt = &pDnode->wrappers[n]; SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
taosMemoryFreeClear(pMgmt->path); taosMemoryFreeClear(pMgmt->path);
} }
if (pDnode->lockfile != NULL) { if (pDnode->data.lockfile != NULL) {
taosUnLockFile(pDnode->lockfile); taosUnLockFile(pDnode->data.lockfile);
taosCloseFile(&pDnode->lockfile); taosCloseFile(&pDnode->data.lockfile);
pDnode->lockfile = NULL; pDnode->data.lockfile = NULL;
} }
taosMemoryFreeClear(pDnode->localEp); taosMemoryFreeClear(pDnode->data.localEp);
taosMemoryFreeClear(pDnode->localFqdn); taosMemoryFreeClear(pDnode->data.localFqdn);
taosMemoryFreeClear(pDnode->firstEp); taosMemoryFreeClear(pDnode->data.firstEp);
taosMemoryFreeClear(pDnode->secondEp); taosMemoryFreeClear(pDnode->data.secondEp);
taosMemoryFreeClear(pDnode->dataDir); taosMemoryFreeClear(pDnode->data.dataDir);
taosMemoryFree(pDnode); taosMemoryFree(pDnode);
dDebug("dnode memory is cleared, data:%p", pDnode); dDebug("dnode memory is cleared, data:%p", pDnode);
} }
@ -82,18 +82,18 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
} }
dndSetStatus(pDnode, DND_STAT_INIT); dndSetStatus(pDnode, DND_STAT_INIT);
dmSetMgmtFp(&pDnode->wrappers[DNODE]); dmSetMgmtFp(&pDnode->wrappers[NODE_BEGIN]);
mmSetMgmtFp(&pDnode->wrappers[MNODE]); mmSetMgmtFp(&pDnode->wrappers[MNODE]);
vmSetMgmtFp(&pDnode->wrappers[VNODES]); vmSetMgmtFp(&pDnode->wrappers[VNODE]);
qmSetMgmtFp(&pDnode->wrappers[QNODE]); qmSetMgmtFp(&pDnode->wrappers[QNODE]);
smSetMgmtFp(&pDnode->wrappers[SNODE]); smSetMgmtFp(&pDnode->wrappers[SNODE]);
bmSetMgmtFp(&pDnode->wrappers[BNODE]); bmSetMgmtFp(&pDnode->wrappers[BNODE]);
for (EDndType n = 0; n < NODE_MAX; ++n) { for (EDndNodeType n = 0; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
snprintf(path, sizeof(path), "%s%s%s", pDnode->dataDir, TD_DIRSEP, pWrapper->name); snprintf(path, sizeof(path), "%s%s%s", pDnode->data.dataDir, TD_DIRSEP, pWrapper->name);
pWrapper->path = strdup(path); pWrapper->path = strdup(path);
pWrapper->shm.id = -1; pWrapper->procShm.id = -1;
pWrapper->pDnode = pDnode; pWrapper->pDnode = pDnode;
pWrapper->ntype = n; pWrapper->ntype = n;
if (pWrapper->path == NULL) { if (pWrapper->path == NULL) {
@ -101,7 +101,7 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
goto _OVER; goto _OVER;
} }
pWrapper->procType = PROC_SINGLE; pWrapper->procType = DND_PROC_SINGLE;
taosInitRWLatch(&pWrapper->latch); taosInitRWLatch(&pWrapper->latch);
} }
@ -134,7 +134,7 @@ _OVER:
void dndClose(SDnode *pDnode) { void dndClose(SDnode *pDnode) {
if (pDnode == NULL) return; if (pDnode == NULL) return;
for (EDndType n = 0; n < NODE_MAX; ++n) { for (EDndNodeType n = 0; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
dndCloseNode(pWrapper); dndCloseNode(pWrapper);
} }
@ -149,7 +149,7 @@ void dndHandleEvent(SDnode *pDnode, EDndEvent event) {
} }
} }
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, EDndType ntype) { SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SMgmtWrapper *pRetWrapper = pWrapper; SMgmtWrapper *pRetWrapper = pWrapper;
@ -170,7 +170,7 @@ int32_t dndMarkWrapper(SMgmtWrapper *pWrapper) {
int32_t code = 0; int32_t code = 0;
taosRLockLatch(&pWrapper->latch); taosRLockLatch(&pWrapper->latch);
if (pWrapper->deployed || (pWrapper->procType == PROC_PARENT && pWrapper->required)) { if (pWrapper->deployed || (pWrapper->procType == DND_PROC_PARENT && pWrapper->required)) {
int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1); int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
dTrace("node:%s, is marked, refCount:%d", pWrapper->name, refCount); dTrace("node:%s, is marked, refCount:%d", pWrapper->name, refCount);
} else { } else {
@ -196,15 +196,6 @@ void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp
pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId; 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) { void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup; SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN); tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);

88
source/dnode/mgmt/main/dndTransport.c
View File

@ -21,7 +21,7 @@
#define INTERNAL_SECRET "_pwd" #define INTERNAL_SECRET "_pwd"
static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) { static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[DNODE]; SMgmtWrapper *pWrapper = &pDnode->wrappers[NODE_BEGIN];
dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet); dmUpdateMnodeEpSet(pWrapper->pMgmt, pEpSet);
} }
@ -53,7 +53,7 @@ static void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpS
int32_t code = -1; int32_t code = -1;
SNodeMsg *pMsg = NULL; SNodeMsg *pMsg = NULL;
NodeMsgFp msgFp = NULL; NodeMsgFp msgFp = NULL;
uint16_t msgType = pRpc->msgType; uint16_t msgType = pRpc->msgType;
if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) { if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) {
dndUpdateMnodeEpSet(pWrapper->pDnode, pEpSet); dndUpdateMnodeEpSet(pWrapper->pDnode, pEpSet);
@ -64,12 +64,12 @@ static void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpS
if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER; if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
if (dndBuildMsg(pMsg, pRpc) != 0) goto _OVER; if (dndBuildMsg(pMsg, pRpc) != 0) goto _OVER;
if (pWrapper->procType == PROC_SINGLE) { if (pWrapper->procType == DND_PROC_SINGLE) {
dTrace("msg:%p, is created, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user); dTrace("msg:%p, is created, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg); code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == PROC_PARENT) { } else if (pWrapper->procType == DND_PROC_PARENT) {
dTrace("msg:%p, is created and put into child queue, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user); dTrace("msg:%p, is created and put into child queue, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle, code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle,
PROC_REQ); PROC_REQ);
} else { } else {
dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user); dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
@ -78,7 +78,7 @@ static void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpS
_OVER: _OVER:
if (code == 0) { if (code == 0) {
if (pWrapper->procType == PROC_PARENT) { if (pWrapper->procType == DND_PROC_PARENT) {
dTrace("msg:%p, is freed in parent process", pMsg); dTrace("msg:%p, is freed in parent process", pMsg);
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
rpcFreeCont(pRpc->pCont); rpcFreeCont(pRpc->pCont);
@ -105,11 +105,11 @@ _OVER:
} }
static void dndProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) { static void dndProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
tmsg_t msgType = pMsg->msgType; tmsg_t msgType = pMsg->msgType;
bool isReq = msgType & 1u; bool isReq = msgType & 1u;
SMsgHandle *pHandle = &pMgmt->msgHandles[TMSG_INDEX(msgType)]; SMsgHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(msgType)];
SMgmtWrapper *pWrapper = pHandle->pWrapper; SMgmtWrapper *pWrapper = pHandle->pNdWrapper;
if (msgType == TDMT_DND_NETWORK_TEST) { if (msgType == TDMT_DND_NETWORK_TEST) {
dTrace("network test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle); dTrace("network test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
@ -159,7 +159,7 @@ static void dndProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
static int32_t dndInitClient(SDnode *pDnode) { static int32_t dndInitClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
SRpcInit rpcInit; SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
@ -178,8 +178,8 @@ static int32_t dndInitClient(SDnode *pDnode) {
taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass); taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
rpcInit.secret = pass; rpcInit.secret = pass;
pMgmt->clientRpc = rpcOpen(&rpcInit); pTrans->clientRpc = rpcOpen(&rpcInit);
if (pMgmt->clientRpc == NULL) { if (pTrans->clientRpc == NULL) {
dError("failed to init dnode rpc client"); dError("failed to init dnode rpc client");
return -1; return -1;
} }
@ -189,17 +189,17 @@ static int32_t dndInitClient(SDnode *pDnode) {
} }
static void dndCleanupClient(SDnode *pDnode) { static void dndCleanupClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
if (pMgmt->clientRpc) { if (pTrans->clientRpc) {
rpcClose(pMgmt->clientRpc); rpcClose(pTrans->clientRpc);
pMgmt->clientRpc = NULL; pTrans->clientRpc = NULL;
dDebug("dnode rpc client is closed"); dDebug("dnode rpc client is closed");
} }
} }
static inline void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) { static inline void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) {
SEpSet epSet = {0}; SEpSet epSet = {0};
SMgmtWrapper *pWrapper = &pDnode->wrappers[DNODE]; SMgmtWrapper *pWrapper = &pDnode->wrappers[NODE_BEGIN];
dmGetMnodeEpSet(pWrapper->pMgmt, &epSet); dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
rpcSendRecv(pDnode->trans.clientRpc, &epSet, pReq, pRsp); rpcSendRecv(pDnode->trans.clientRpc, &epSet, pReq, pRsp);
} }
@ -263,11 +263,11 @@ static int32_t dndRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *spi, ch
} }
static int32_t dndInitServer(SDnode *pDnode) { static int32_t dndInitServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
SRpcInit rpcInit; SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localPort = pDnode->serverPort; rpcInit.localPort = pDnode->data.serverPort;
rpcInit.label = "DND"; rpcInit.label = "DND";
rpcInit.numOfThreads = tsNumOfRpcThreads; rpcInit.numOfThreads = tsNumOfRpcThreads;
rpcInit.cfp = (RpcCfp)dndProcessMsg; rpcInit.cfp = (RpcCfp)dndProcessMsg;
@ -277,8 +277,8 @@ static int32_t dndInitServer(SDnode *pDnode) {
rpcInit.afp = (RpcAfp)dndRetrieveUserAuthInfo; rpcInit.afp = (RpcAfp)dndRetrieveUserAuthInfo;
rpcInit.parent = pDnode; rpcInit.parent = pDnode;
pMgmt->serverRpc = rpcOpen(&rpcInit); pTrans->serverRpc = rpcOpen(&rpcInit);
if (pMgmt->serverRpc == NULL) { if (pTrans->serverRpc == NULL) {
dError("failed to init dnode rpc server"); dError("failed to init dnode rpc server");
return -1; return -1;
} }
@ -288,10 +288,10 @@ static int32_t dndInitServer(SDnode *pDnode) {
} }
static void dndCleanupServer(SDnode *pDnode) { static void dndCleanupServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
if (pMgmt->serverRpc) { if (pTrans->serverRpc) {
rpcClose(pMgmt->serverRpc); rpcClose(pTrans->serverRpc);
pMgmt->serverRpc = NULL; pTrans->serverRpc = NULL;
dDebug("dnode rpc server is closed"); dDebug("dnode rpc server is closed");
} }
} }
@ -308,9 +308,9 @@ void dndCleanupTrans(SDnode *pDnode) {
} }
int32_t dndInitMsgHandle(SDnode *pDnode) { int32_t dndInitMsgHandle(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
for (EDndType n = 0; n < NODE_MAX; ++n) { for (EDndNodeType n = 0; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n]; SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
for (int32_t msgIndex = 0; msgIndex < TDMT_MAX; ++msgIndex) { for (int32_t msgIndex = 0; msgIndex < TDMT_MAX; ++msgIndex) {
@ -318,7 +318,7 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
int8_t vgId = pWrapper->msgVgIds[msgIndex]; int8_t vgId = pWrapper->msgVgIds[msgIndex];
if (msgFp == NULL) continue; if (msgFp == NULL) continue;
SMsgHandle *pHandle = &pMgmt->msgHandles[msgIndex]; SMsgHandle *pHandle = &pTrans->msgHandles[msgIndex];
if (vgId == QNODE_HANDLE) { if (vgId == QNODE_HANDLE) {
if (pHandle->pQndWrapper != NULL) { if (pHandle->pQndWrapper != NULL) {
dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]); dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]);
@ -332,11 +332,11 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
} }
pHandle->pMndWrapper = pWrapper; pHandle->pMndWrapper = pWrapper;
} else { } else {
if (pHandle->pWrapper != NULL) { if (pHandle->pNdWrapper != NULL) {
dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]); dError("msg:%s has multiple process nodes", tMsgInfo[msgIndex]);
return -1; return -1;
} }
pHandle->pWrapper = pWrapper; pHandle->pNdWrapper = pWrapper;
} }
} }
} }
@ -344,13 +344,13 @@ int32_t dndInitMsgHandle(SDnode *pDnode) {
return 0; return 0;
} }
static int32_t dndSendRpcReq(STransMgmt *pMgmt, const SEpSet *pEpSet, SRpcMsg *pReq) { static int32_t dndSendRpcReq(SDnodeTrans *pTrans, const SEpSet *pEpSet, SRpcMsg *pReq) {
if (pMgmt->clientRpc == NULL) { if (pTrans->clientRpc == NULL) {
terrno = TSDB_CODE_NODE_OFFLINE; terrno = TSDB_CODE_NODE_OFFLINE;
return -1; return -1;
} }
rpcSendRequest(pMgmt->clientRpc, pEpSet, pReq, NULL); rpcSendRequest(pTrans->clientRpc, pEpSet, pReq, NULL);
return 0; return 0;
} }
@ -369,7 +369,7 @@ static int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg
return -1; return -1;
} }
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {
return dndSendRpcReq(&pWrapper->pDnode->trans, pEpSet, pReq); return dndSendRpcReq(&pWrapper->pDnode->trans, pEpSet, pReq);
} else { } else {
char *pHead = taosMemoryMalloc(sizeof(SRpcMsg) + sizeof(SEpSet)); char *pHead = taosMemoryMalloc(sizeof(SRpcMsg) + sizeof(SEpSet));
@ -380,7 +380,7 @@ static int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg
memcpy(pHead, pReq, sizeof(SRpcMsg)); memcpy(pHead, pReq, sizeof(SRpcMsg));
memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet)); memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet));
taosProcPutToParentQ(pWrapper->pProc, pHead, sizeof(SRpcMsg) + sizeof(SEpSet), pReq->pCont, pReq->contLen, taosProcPutToParentQ(pWrapper->procObj, pHead, sizeof(SRpcMsg) + sizeof(SEpSet), pReq->pCont, pReq->contLen,
PROC_REQ); PROC_REQ);
taosMemoryFree(pHead); taosMemoryFree(pHead);
return 0; return 0;
@ -388,27 +388,27 @@ static int32_t dndSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg
} }
static void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) { static void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {
dndSendRpcRsp(pWrapper, pRsp); dndSendRpcRsp(pWrapper, pRsp);
} else { } else {
taosProcPutToParentQ(pWrapper->pProc, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_RSP); taosProcPutToParentQ(pWrapper->procObj, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_RSP);
} }
} }
static void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) { static void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {
rpcRegisterBrokenLinkArg(pMsg); rpcRegisterBrokenLinkArg(pMsg);
} else { } else {
taosProcPutToParentQ(pWrapper->pProc, pMsg, sizeof(SRpcMsg), pMsg->pCont, pMsg->contLen, PROC_REGIST); taosProcPutToParentQ(pWrapper->procObj, pMsg, sizeof(SRpcMsg), pMsg->pCont, pMsg->contLen, PROC_REGIST);
} }
} }
static void dndReleaseHandle(SMgmtWrapper *pWrapper, void *handle, int8_t type) { static void dndReleaseHandle(SMgmtWrapper *pWrapper, void *handle, int8_t type) {
if (pWrapper->procType != PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {
rpcReleaseHandle(handle, type); rpcReleaseHandle(handle, type);
} else { } else {
SRpcMsg msg = {.handle = handle, .code = type}; SRpcMsg msg = {.handle = handle, .code = type};
taosProcPutToParentQ(pWrapper->pProc, &msg, sizeof(SRpcMsg), NULL, 0, PROC_RELEASE); taosProcPutToParentQ(pWrapper->procObj, &msg, sizeof(SRpcMsg), NULL, 0, PROC_RELEASE);
} }
} }
@ -456,7 +456,7 @@ static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
rpcRegisterBrokenLinkArg(pMsg); rpcRegisterBrokenLinkArg(pMsg);
break; break;
case PROC_RELEASE: case PROC_RELEASE:
taosProcRemoveHandle(pWrapper->pProc, pMsg->handle); taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code); rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
rpcFreeCont(pCont); rpcFreeCont(pCont);
break; break;
@ -464,7 +464,7 @@ static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
dndSendRpcReq(&pWrapper->pDnode->trans, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg); dndSendRpcReq(&pWrapper->pDnode->trans, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
break; break;
case PROC_RSP: case PROC_RSP:
taosProcRemoveHandle(pWrapper->pProc, pMsg->handle); taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
dndSendRpcRsp(pWrapper, pMsg); dndSendRpcRsp(pWrapper, pMsg);
break; break;
default: default:
@ -484,7 +484,7 @@ SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree, .parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont, .parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont, .parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->shm, .shm = pWrapper->procShm,
.parent = pWrapper, .parent = pWrapper,
.name = pWrapper->name}; .name = pWrapper->name};
return cfg; return cfg;

8
source/dnode/mgmt/mm/mmHandle.c
View File

@ -56,7 +56,7 @@ int32_t mmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (createReq.replica <= 1 || createReq.dnodeId != pDnode->dnodeId) { if (createReq.replica <= 1 || createReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to create mnode since %s", terrstr()); dError("failed to create mnode since %s", terrstr());
return -1; return -1;
@ -75,7 +75,7 @@ int32_t mmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (dropReq.dnodeId != pDnode->dnodeId) { if (dropReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop mnode since %s", terrstr()); dError("failed to drop mnode since %s", terrstr());
return -1; return -1;
@ -95,9 +95,9 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (alterReq.dnodeId != pDnode->dnodeId) { if (alterReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to alter mnode since %s, dnodeId:%d input:%d", terrstr(), pDnode->dnodeId, alterReq.dnodeId); dError("failed to alter mnode since %s, dnodeId:%d input:%d", terrstr(), pDnode->data.dnodeId, alterReq.dnodeId);
return -1; return -1;
} else { } else {
return mmAlter(pMgmt, &alterReq); return mmAlter(pMgmt, &alterReq);

20
source/dnode/mgmt/mm/mmInt.c
View File

@ -18,9 +18,9 @@
#include "wal.h" #include "wal.h"
static bool mmDeployRequired(SDnode *pDnode) { static bool mmDeployRequired(SDnode *pDnode) {
if (pDnode->dnodeId > 0) return false; if (pDnode->data.dnodeId > 0) return false;
if (pDnode->clusterId > 0) return false; if (pDnode->data.clusterId > 0) return false;
if (strcmp(pDnode->localEp, pDnode->firstEp) != 0) return false; if (strcmp(pDnode->data.localEp, pDnode->data.firstEp) != 0) return false;
return true; return true;
} }
@ -53,8 +53,8 @@ static void mmBuildOptionForDeploy(SMnodeMgmt *pMgmt, SMnodeOpt *pOption) {
pOption->selfIndex = 0; pOption->selfIndex = 0;
SReplica *pReplica = &pOption->replicas[0]; SReplica *pReplica = &pOption->replicas[0];
pReplica->id = 1; pReplica->id = 1;
pReplica->port = pMgmt->pDnode->serverPort; pReplica->port = pMgmt->pDnode->data.serverPort;
tstrncpy(pReplica->fqdn, pMgmt->pDnode->localFqdn, TSDB_FQDN_LEN); tstrncpy(pReplica->fqdn, pMgmt->pDnode->data.localFqdn, TSDB_FQDN_LEN);
pOption->deploy = true; pOption->deploy = true;
pMgmt->selfIndex = pOption->selfIndex; pMgmt->selfIndex = pOption->selfIndex;
@ -80,7 +80,7 @@ static int32_t mmBuildOptionFromReq(SMnodeMgmt *pMgmt, SMnodeOpt *pOption, SDCre
pReplica->id = pCreate->replicas[i].id; pReplica->id = pCreate->replicas[i].id;
pReplica->port = pCreate->replicas[i].port; pReplica->port = pCreate->replicas[i].port;
memcpy(pReplica->fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN); memcpy(pReplica->fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
if (pReplica->id == pMgmt->pDnode->dnodeId) { if (pReplica->id == pMgmt->pDnode->data.dnodeId) {
pOption->selfIndex = i; pOption->selfIndex = i;
} }
} }
@ -112,8 +112,8 @@ static int32_t mmOpenImp(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pReq) {
if (!deployed) { if (!deployed) {
dInfo("mnode start to deploy"); dInfo("mnode start to deploy");
if (pMgmt->pWrapper->procType == PROC_CHILD) { if (pMgmt->pWrapper->procType == DND_PROC_CHILD) {
pMgmt->pDnode->dnodeId = 1; pMgmt->pDnode->data.dnodeId = 1;
} }
mmBuildOptionForDeploy(pMgmt, &option); mmBuildOptionForDeploy(pMgmt, &option);
} else { } else {
@ -230,8 +230,8 @@ void mmSetMgmtFp(SMgmtWrapper *pWrapper) {
mgmtFp.openFp = mmOpen; mgmtFp.openFp = mmOpen;
mgmtFp.closeFp = mmClose; mgmtFp.closeFp = mmClose;
mgmtFp.startFp = mmStart; mgmtFp.startFp = mmStart;
mgmtFp.createMsgFp = mmProcessCreateReq; mgmtFp.createFp = mmProcessCreateReq;
mgmtFp.dropMsgFp = mmProcessDropReq; mgmtFp.dropFp = mmProcessDropReq;
mgmtFp.requiredFp = mmRequire; mgmtFp.requiredFp = mmRequire;
mmInitMsgHandle(pWrapper); mmInitMsgHandle(pWrapper);

4
source/dnode/mgmt/qm/qmHandle.c
View File

@ -53,7 +53,7 @@ int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (createReq.dnodeId != pDnode->dnodeId) { if (createReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to create qnode since %s", terrstr()); dError("failed to create qnode since %s", terrstr());
return -1; return -1;
@ -72,7 +72,7 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (dropReq.dnodeId != pDnode->dnodeId) { if (dropReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop qnode since %s", terrstr()); dError("failed to drop qnode since %s", terrstr());
return -1; return -1;

4
source/dnode/mgmt/qm/qmInt.c
View File

@ -116,8 +116,8 @@ void qmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = qmOpen; mgmtFp.openFp = qmOpen;
mgmtFp.closeFp = qmClose; mgmtFp.closeFp = qmClose;
mgmtFp.createMsgFp = qmProcessCreateReq; mgmtFp.createFp = qmProcessCreateReq;
mgmtFp.dropMsgFp = qmProcessDropReq; mgmtFp.dropFp = qmProcessDropReq;
mgmtFp.requiredFp = qmRequire; mgmtFp.requiredFp = qmRequire;
qmInitMsgHandle(pWrapper); qmInitMsgHandle(pWrapper);

4
source/dnode/mgmt/sm/smHandle.c
View File

@ -53,7 +53,7 @@ int32_t smProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (createReq.dnodeId != pDnode->dnodeId) { if (createReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to create snode since %s", terrstr()); dError("failed to create snode since %s", terrstr());
return -1; return -1;
@ -72,7 +72,7 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return -1; return -1;
} }
if (dropReq.dnodeId != pDnode->dnodeId) { if (dropReq.dnodeId != pDnode->data.dnodeId) {
terrno = TSDB_CODE_INVALID_OPTION; terrno = TSDB_CODE_INVALID_OPTION;
dError("failed to drop snode since %s", terrstr()); dError("failed to drop snode since %s", terrstr());
return -1; return -1;

4
source/dnode/mgmt/sm/smInt.c
View File

@ -113,8 +113,8 @@ void smSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0}; SMgmtFp mgmtFp = {0};
mgmtFp.openFp = smOpen; mgmtFp.openFp = smOpen;
mgmtFp.closeFp = smClose; mgmtFp.closeFp = smClose;
mgmtFp.createMsgFp = smProcessCreateReq; mgmtFp.createFp = smProcessCreateReq;
mgmtFp.dropMsgFp = smProcessDropReq; mgmtFp.dropFp = smProcessDropReq;
mgmtFp.requiredFp = smRequire; mgmtFp.requiredFp = smRequire;
smInitMsgHandle(pWrapper); smInitMsgHandle(pWrapper);

8
source/dnode/mgmt/vm/vmInt.c
View File

@ -278,11 +278,11 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
taosInitRWLatch(&pMgmt->latch); taosInitRWLatch(&pMgmt->latch);
SDiskCfg dCfg = {0}; SDiskCfg dCfg = {0};
tstrncpy(dCfg.dir, pDnode->dataDir, TSDB_FILENAME_LEN); tstrncpy(dCfg.dir, pDnode->data.dataDir, TSDB_FILENAME_LEN);
dCfg.level = 0; dCfg.level = 0;
dCfg.primary = 1; dCfg.primary = 1;
SDiskCfg *pDisks = pDnode->disks; SDiskCfg *pDisks = pDnode->data.disks;
int32_t numOfDisks = pDnode->numOfDisks; int32_t numOfDisks = pDnode->data.numOfDisks;
if (numOfDisks <= 0 || pDisks == NULL) { if (numOfDisks <= 0 || pDisks == NULL) {
pDisks = &dCfg; pDisks = &dCfg;
numOfDisks = 1; numOfDisks = 1;
@ -329,7 +329,7 @@ _OVER:
static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) { static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) {
SDnode *pDnode = pWrapper->pDnode; SDnode *pDnode = pWrapper->pDnode;
*required = pDnode->numOfSupportVnodes > 0; *required = pDnode->data.supportVnodes > 0;
return 0; return 0;
} }

4
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -354,7 +354,7 @@ static int32_t mndProcessStatusReq(SNodeMsg *pReq) {
int64_t curMs = taosGetTimestampMs(); int64_t curMs = taosGetTimestampMs();
bool online = mndIsDnodeOnline(pMnode, pDnode, curMs); bool online = mndIsDnodeOnline(pMnode, pDnode, curMs);
bool dnodeChanged = (statusReq.dver != sdbGetTableVer(pMnode->pSdb, SDB_DNODE)); bool dnodeChanged = (statusReq.dnodeVer != sdbGetTableVer(pMnode->pSdb, SDB_DNODE));
bool reboot = (pDnode->rebootTime != statusReq.rebootTime); bool reboot = (pDnode->rebootTime != statusReq.rebootTime);
bool needCheck = !online || dnodeChanged || reboot; bool needCheck = !online || dnodeChanged || reboot;
@ -405,7 +405,7 @@ static int32_t mndProcessStatusReq(SNodeMsg *pReq) {
pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes; pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes;
SStatusRsp statusRsp = {0}; SStatusRsp statusRsp = {0};
statusRsp.dver = sdbGetTableVer(pMnode->pSdb, SDB_DNODE); statusRsp.dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE);
statusRsp.dnodeCfg.dnodeId = pDnode->id; statusRsp.dnodeCfg.dnodeId = pDnode->id;
statusRsp.dnodeCfg.clusterId = pMnode->clusterId; statusRsp.dnodeCfg.clusterId = pMnode->clusterId;
statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp)); statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp));