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

refactor: mnode sync

This commit is contained in:
Shengliang Guan 2022-06-14 10:58:36 +08:00
parent 01521a2541
commit a3f1273640
8 changed files with 27 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/dnode/mnode/mnode.h
View File

@ -32,9 +32,7 @@ typedef struct {
int32_t dnodeId; int32_t dnodeId;
bool standby; bool standby;
bool deploy; bool deploy;
int8_t replica; SReplica replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
SMsgCb msgCb; SMsgCb msgCb;
} SMnodeOpt; } SMnodeOpt;

4
source/dnode/mgmt/mgmt_mnode/src/mmFile.c
View File

@ -86,12 +86,14 @@ int32_t mmReadFile(SMnodeMgmt *pMgmt, SReplica *pReplica, bool *pDeployed) {
} }
code = 0; code = 0;
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
_OVER: _OVER:
if (content != NULL) taosMemoryFree(content); if (content != NULL) taosMemoryFree(content);
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile); if (pFile != NULL) taosCloseFile(&pFile);
if (code == 0) {
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
}
terrno = code; terrno = code;
return code; return code;

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

@ -126,7 +126,7 @@ int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
SArray *mmGetMsgHandles() { SArray *mmGetMsgHandles() {
int32_t code = -1; int32_t code = -1;
SArray *pArray = taosArrayInit(64, sizeof(SMgmtHandle)); SArray *pArray = taosArrayInit(128, sizeof(SMgmtHandle));
if (pArray == NULL) goto _OVER; if (pArray == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@ -43,13 +43,9 @@ static void mmBuildOptionForDeploy(SMnodeMgmt *pMgmt, const SMgmtInputOpt *pInpu
pOption->deploy = true; pOption->deploy = true;
pOption->msgCb = pMgmt->msgCb; pOption->msgCb = pMgmt->msgCb;
pOption->dnodeId = pMgmt->pData->dnodeId; pOption->dnodeId = pMgmt->pData->dnodeId;
pOption->replica = 1; pOption->replica.id = 1;
pOption->selfIndex = 0; pOption->replica.port = tsServerPort;
tstrncpy(pOption->replica.fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
SReplica *pReplica = &pOption->replicas[0];
pReplica->id = 1;
pReplica->port = tsServerPort;
tstrncpy(pReplica->fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
} }
static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, const SReplica *pReplica, SMnodeOpt *pOption) { static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, const SReplica *pReplica, SMnodeOpt *pOption) {
@ -57,12 +53,9 @@ static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, const SReplica *pReplica, SM
pOption->deploy = false; pOption->deploy = false;
pOption->msgCb = pMgmt->msgCb; pOption->msgCb = pMgmt->msgCb;
pOption->dnodeId = pMgmt->pData->dnodeId; pOption->dnodeId = pMgmt->pData->dnodeId;
if (pReplica->id > 0) { if (pReplica->id > 0) {
pOption->standby = true; pOption->standby = true;
pOption->replica = 1; pOption->replica = *pReplica;
pOption->selfIndex = 0;
memcpy(&pOption->replicas[0], pReplica, sizeof(SReplica));
} }
} }

14
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -76,11 +76,12 @@ typedef struct {
} STelemMgmt; } STelemMgmt;
typedef struct { typedef struct {
sem_t syncSem; sem_t syncSem;
int64_t sync; int64_t sync;
bool standby; bool standby;
int32_t errCode; SReplica replica;
int32_t transId; int32_t errCode;
int32_t transId;
} SSyncMgmt; } SSyncMgmt;
typedef struct { typedef struct {
@ -98,9 +99,6 @@ typedef struct SMnode {
bool stopped; bool stopped;
bool restored; bool restored;
bool deploy; bool deploy;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
char *path; char *path;
int64_t checkTime; int64_t checkTime;
SSdb *pSdb; SSdb *pSdb;

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

@ -95,8 +95,8 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
dnodeObj.id = 1; dnodeObj.id = 1;
dnodeObj.createdTime = taosGetTimestampMs(); dnodeObj.createdTime = taosGetTimestampMs();
dnodeObj.updateTime = dnodeObj.createdTime; dnodeObj.updateTime = dnodeObj.createdTime;
dnodeObj.port = pMnode->replicas[0].port; dnodeObj.port = tsServerPort;
memcpy(&dnodeObj.fqdn, pMnode->replicas[0].fqdn, TSDB_FQDN_LEN); memcpy(&dnodeObj.fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port); snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL);

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

@ -289,11 +289,9 @@ static int32_t mndExecSteps(SMnode *pMnode) {
} }
static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) { static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
pMnode->replica = pOption->replica;
pMnode->selfIndex = pOption->selfIndex;
memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
pMnode->msgCb = pOption->msgCb; pMnode->msgCb = pOption->msgCb;
pMnode->selfDnodeId = pOption->dnodeId; pMnode->selfDnodeId = pOption->dnodeId;
pMnode->syncMgmt.replica = pOption->replica;
pMnode->syncMgmt.standby = pOption->standby; pMnode->syncMgmt.standby = pOption->standby;
} }

18
source/dnode/mnode/impl/src/mndSync.c
View File

@ -188,15 +188,15 @@ int32_t mndInitSync(SMnode *pMnode) {
syncInfo.isStandBy = pMgmt->standby; syncInfo.isStandBy = pMgmt->standby;
syncInfo.snapshotEnable = true; syncInfo.snapshotEnable = true;
SSyncCfg *pCfg = &syncInfo.syncCfg; mInfo("start to open mnode sync, standby:%d", pMgmt->standby);
pCfg->replicaNum = pMnode->replica; if (pMgmt->standby || pMgmt->replica.id > 0) {
pCfg->myIndex = pMnode->selfIndex; SSyncCfg *pCfg = &syncInfo.syncCfg;
mInfo("start to open mnode sync, replica:%d myindex:%d standby:%d", pCfg->replicaNum, pCfg->myIndex, pMgmt->standby); pCfg->replicaNum = 1;
for (int32_t i = 0; i < pMnode->replica; ++i) { pCfg->myIndex = 0;
SNodeInfo *pNode = &pCfg->nodeInfo[i]; SNodeInfo *pNode = &pCfg->nodeInfo[0];
tstrncpy(pNode->nodeFqdn, pMnode->replicas[i].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pMgmt->replica.fqdn, sizeof(pNode->nodeFqdn));
pNode->nodePort = pMnode->replicas[i].port; pNode->nodePort = pMgmt->replica.port;
mInfo("index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort); mInfo("fqdn:%s port:%u", pNode->nodeFqdn, pNode->nodePort);
} }
tsem_init(&pMgmt->syncSem, 0, 0); tsem_init(&pMgmt->syncSem, 0, 0);