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

refactor: sync integrate into mnode

This commit is contained in:
Shengliang Guan 2022-05-22 16:42:44 +08:00
parent e06f9fd05c
commit 32d48f3b8f
3 changed files with 34 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -46,9 +46,8 @@ static void mndCloseWal(SMnode *pMnode) {
} }
static int32_t mndRestoreWal(SMnode *pMnode) { static int32_t mndRestoreWal(SMnode *pMnode) {
// do nothing
// do nothing return 0;
return 0;
#if 0 #if 0
@ -122,7 +121,6 @@ _OVER:
return code; return code;
#endif #endif
} }
int32_t mndSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) { return tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg); } int32_t mndSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) { return tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg); }
@ -132,7 +130,7 @@ int32_t mndSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { return tmsgSendReq
void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
SyncIndex beginIndex = SYNC_INDEX_INVALID; SyncIndex beginIndex = SYNC_INDEX_INVALID;
if (pFsm->FpGetSnapshot != NULL) { if (pFsm->FpGetSnapshot != NULL) {
SSnapshot snapshot; SSnapshot snapshot = {0};
pFsm->FpGetSnapshot(pFsm, &snapshot); pFsm->FpGetSnapshot(pFsm, &snapshot);
beginIndex = snapshot.lastApplyIndex; beginIndex = snapshot.lastApplyIndex;
} }
@ -141,8 +139,9 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
mndProcessApplyMsg((SRpcMsg*)pMsg); SRpcMsg *pApplyMsg = (SRpcMsg *)pMsg;
//mmPutNodeMsgToApplyQueue(pMnode->pWrapper->pMgmt, pMsg); pApplyMsg->info.node = pFsm->data;
mndProcessApplyMsg(pApplyMsg);
if (cbMeta.state == TAOS_SYNC_STATE_LEADER) { if (cbMeta.state == TAOS_SYNC_STATE_LEADER) {
tsem_post(&pMgmt->syncSem); tsem_post(&pMgmt->syncSem);
@ -182,19 +181,16 @@ int32_t mndInitSync(SMnode *pMnode) {
return -1; return -1;
} }
if (pMnode->selfId == 1) {
pMgmt->state = TAOS_SYNC_STATE_LEADER;
}
SSyncInfo syncInfo = {.vgId = 1}; SSyncInfo syncInfo = {.vgId = 1};
SSyncCfg *pCfg = &(syncInfo.syncCfg); SSyncCfg *pCfg = &(syncInfo.syncCfg);
pCfg->replicaNum = pMnode->replica; pCfg->replicaNum = pMnode->replica;
pCfg->myIndex = pMnode->selfIndex; pCfg->myIndex = pMnode->selfIndex;
for (int i = 0; i < pMnode->replica; ++i) { for (int32_t i = 0; i < pMnode->replica; ++i) {
snprintf((pCfg->nodeInfo)->nodeFqdn, sizeof((pCfg->nodeInfo)->nodeFqdn), "%s", (pMnode->replicas)[i].fqdn); SNodeInfo *pNodeInfo = &pCfg->nodeInfo[i];
(pCfg->nodeInfo)->nodePort = (pMnode->replicas)[i].port; tstrncpy(pNodeInfo->nodeFqdn, pMnode->replicas[i].fqdn, sizeof(pNodeInfo->nodeFqdn));
pNodeInfo->nodePort = pMnode->replicas[i].port;
} }
snprintf(syncInfo.path, sizeof(syncInfo.path), "%s/sync", pMnode->path); snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", pMnode->path, TD_DIRSEP);
syncInfo.pWal = pMnode->syncMgmt.pWal; syncInfo.pWal = pMnode->syncMgmt.pWal;
syncInfo.pFsm = mndSyncMakeFsm(pMnode); syncInfo.pFsm = mndSyncMakeFsm(pMnode);
syncInfo.FpSendMsg = mndSyncSendMsg; syncInfo.FpSendMsg = mndSyncSendMsg;
@ -242,31 +238,38 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw) {
#else #else
if (pMnode->replica == 1) return 0;
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
pMgmt->errCode = 0; pMgmt->errCode = 0;
//SSyncBuffer buf = {.data = pRaw, .len = sdbGetRawTotalSize(pRaw)}; SRpcMsg rsp = {0};
rsp.code = TDMT_MND_APPLY_MSG;
SRpcMsg rpcMsg; rsp.contLen = sdbGetRawTotalSize(pRaw);
rpcMsg.code = TDMT_MND_APPLY_MSG; rsp.pCont = rpcMallocCont(rsp.contLen);
rpcMsg.contLen = sdbGetRawTotalSize(pRaw); memcpy(rsp.pCont, pRaw, rsp.contLen);
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
memcpy(rpcMsg.pCont, pRaw, rpcMsg.contLen);
bool isWeak = false; bool isWeak = false;
int32_t code = syncPropose(pMgmt->sync, &rpcMsg, isWeak); int32_t code = syncPropose(pMgmt->sync, &rsp, isWeak);
if (code == 0) {
tsem_wait(&pMgmt->syncSem);
} else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
terrno = TSDB_CODE_APP_NOT_READY;
mError("failed to propose raw:%p since not leader", pRaw);
return -1;
} else if (code == TAOS_SYNC_PROPOSE_OTHER_ERROR) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
mError("failed to propose raw:%p since sync internal error", pRaw);
} else {
assert(0);
}
if (code != 0) return code; if (code != 0) return code;
tsem_wait(&pMgmt->syncSem); tsem_wait(&pMgmt->syncSem);
return pMgmt->errCode; return pMgmt->errCode;
#endif #endif
} }
bool mndIsMaster(SMnode *pMnode) { bool mndIsMaster(SMnode *pMnode) {
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
pMgmt->state = syncGetMyRole(pMgmt->sync);
return pMgmt->state == TAOS_SYNC_STATE_LEADER; return pMgmt->state == TAOS_SYNC_STATE_LEADER;
} }

5
source/dnode/mnode/impl/src/mnode.c
View File

@ -336,8 +336,9 @@ int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption) {
return 0; return 0;
} }
int32_t mndStart(SMnode *pMnode) { int32_t mndStart(SMnode *pMnode) {
syncStart(pMnode->syncMgmt.sync); syncSetMsgCb(pMnode->syncMgmt.sync, &pMnode->msgCb);
syncStart(pMnode->syncMgmt.sync);
return mndInitTimer(pMnode); return mndInitTimer(pMnode);
} }

10
source/dnode/mnode/sdb/src/sdb.c
View File

@ -31,11 +31,9 @@ SSdb *sdbInit(SSdbOpt *pOption) {
char path[PATH_MAX + 100] = {0}; char path[PATH_MAX + 100] = {0};
snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP); snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP);
pSdb->currDir = strdup(path); pSdb->currDir = strdup(path);
snprintf(path, sizeof(path), "%s%ssync", pOption->path, TD_DIRSEP);
pSdb->syncDir = strdup(path);
snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP); snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP);
pSdb->tmpDir = strdup(path); pSdb->tmpDir = strdup(path);
if (pSdb->currDir == NULL || pSdb->currDir == NULL || pSdb->currDir == NULL) { if (pSdb->currDir == NULL || pSdb->tmpDir == NULL) {
sdbCleanup(pSdb); sdbCleanup(pSdb);
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("failed to init sdb since %s", terrstr()); mError("failed to init sdb since %s", terrstr());
@ -149,12 +147,6 @@ static int32_t sdbCreateDir(SSdb *pSdb) {
return -1; return -1;
} }
if (taosMkDir(pSdb->syncDir) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to create dir:%s since %s", pSdb->syncDir, terrstr());
return -1;
}
if (taosMkDir(pSdb->tmpDir) != 0) { if (taosMkDir(pSdb->tmpDir) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to create dir:%s since %s", pSdb->tmpDir, terrstr()); mError("failed to create dir:%s since %s", pSdb->tmpDir, terrstr());