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

fix: handle error code

This commit is contained in:
Shengliang Guan 2024-08-20 09:54:26 +08:00
parent 660bfde593
commit 1f1a2d9f6c
5 changed files with 60 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/sync/inc/syncRaftCfg.h
View File

@ -24,7 +24,6 @@ extern "C" {
int32_t syncWriteCfgFile(SSyncNode *pNode); int32_t syncWriteCfgFile(SSyncNode *pNode);
int32_t syncReadCfgFile(SSyncNode *pNode); int32_t syncReadCfgFile(SSyncNode *pNode);
int32_t syncAddCfgIndex(SSyncNode *pNode, SyncIndex cfgIndex);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/libs/sync/inc/syncUtil.h
View File

@ -107,7 +107,7 @@ void syncLogRecvAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMs
void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s); void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s);
void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, int32_t voteGranted, const char* s); void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, int32_t voteGranted, const char* s);
void syncLogSendRequestVote(SSyncNode* pNode, const SyncRequestVote* pMsg, const char* s); void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s);
void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s); void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s);
void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s); void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s);

26
source/libs/sync/src/syncMain.c
View File

@ -50,7 +50,7 @@ static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer);
static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer); static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer);
static int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg); static int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg);
static bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config); static bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config);
static void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex); static int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex);
static bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg); static bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg);
static bool syncNodeCanChange(SSyncNode* pSyncNode); static bool syncNodeCanChange(SSyncNode* pSyncNode);
@ -182,7 +182,12 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
} }
TAOS_CHECK_RETURN(syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg)); TAOS_CHECK_RETURN(syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg));
syncNodeDoConfigChange(pSyncNode, pNewCfg, pNewCfg->lastIndex);
if (syncNodeDoConfigChange(pSyncNode, pNewCfg, pNewCfg->lastIndex) != 0) {
code = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
sError("vgId:%d, failed to reconfig since do change error", pSyncNode->vgId);
TAOS_RETURN(code);
}
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER || pSyncNode->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER || pSyncNode->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) {
// TODO check return value // TODO check return value
@ -1015,7 +1020,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
if (!taosDirExist((char*)(pSyncInfo->path))) { if (!taosDirExist((char*)(pSyncInfo->path))) {
if (taosMkDir(pSyncInfo->path) != 0) { if (taosMkDir(pSyncInfo->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
sError("failed to create dir:%s since %s", pSyncInfo->path, terrstr()); sError("vgId:%d, failed to create dir:%s since %s", pSyncInfo->vgId, pSyncInfo->path, terrstr());
goto _error; goto _error;
} }
} }
@ -1766,11 +1771,11 @@ static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg
return false; return false;
} }
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) { int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
SSyncCfg oldConfig = pSyncNode->raftCfg.cfg; SSyncCfg oldConfig = pSyncNode->raftCfg.cfg;
if (!syncIsConfigChanged(&oldConfig, pNewConfig)) { if (!syncIsConfigChanged(&oldConfig, pNewConfig)) {
sInfo("vgId:1, sync not reconfig since not changed"); sInfo("vgId:1, sync not reconfig since not changed");
return; return 0;
} }
pSyncNode->raftCfg.cfg = *pNewConfig; pSyncNode->raftCfg.cfg = *pNewConfig;
@ -1809,7 +1814,15 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
} }
// add last config index // add last config index
(void)syncAddCfgIndex(pSyncNode, lastConfigChangeIndex); SRaftCfg* pCfg = &pSyncNode->raftCfg;
if (pCfg->configIndexCount >= MAX_CONFIG_INDEX_COUNT) {
sNError(pSyncNode, "failed to add cfg index:%d since out of range", pCfg->configIndexCount);
terrno = TSDB_CODE_OUT_OF_RANGE;
return -1;
}
pCfg->configIndexArr[pCfg->configIndexCount] = lastConfigChangeIndex;
pCfg->configIndexCount++;
if (IamInNew) { if (IamInNew) {
//----------------------------------------- //-----------------------------------------
@ -1924,6 +1937,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
_END: _END:
// log end config change // log end config change
sNInfo(pSyncNode, "end do config change, from %d to %d", oldConfig.totalReplicaNum, pNewConfig->totalReplicaNum); sNInfo(pSyncNode, "end do config change, from %d to %d", oldConfig.totalReplicaNum, pNewConfig->totalReplicaNum);
return 0;
} }
// raft state change -------------- // raft state change --------------

42
source/libs/sync/src/syncRaftCfg.c
View File

@ -18,7 +18,7 @@
#include "syncUtil.h" #include "syncUtil.h"
#include "tjson.h" #include "tjson.h"
const char *syncRoleToStr(ESyncRole role) { static const char *syncRoleToStr(ESyncRole role) {
switch (role) { switch (role) {
case TAOS_SYNC_ROLE_VOTER: case TAOS_SYNC_ROLE_VOTER:
return "true"; return "true";
@ -29,16 +29,15 @@ const char *syncRoleToStr(ESyncRole role) {
} }
} }
const ESyncRole syncStrToRole(char *str) { static const ESyncRole syncStrToRole(char *str) {
if (strcmp(str, "true") == 0) { if (strcmp(str, "true") == 0) {
return TAOS_SYNC_ROLE_VOTER; return TAOS_SYNC_ROLE_VOTER;
} } else if (strcmp(str, "false") == 0) {
if (strcmp(str, "false") == 0) {
return TAOS_SYNC_ROLE_LEARNER; return TAOS_SYNC_ROLE_LEARNER;
} } else {
return TAOS_SYNC_ROLE_ERROR; return TAOS_SYNC_ROLE_ERROR;
} }
}
static int32_t syncEncodeSyncCfg(const void *pObj, SJson *pJson) { static int32_t syncEncodeSyncCfg(const void *pObj, SJson *pJson) {
SSyncCfg *pCfg = (SSyncCfg *)pObj; SSyncCfg *pCfg = (SSyncCfg *)pObj;
@ -52,10 +51,12 @@ static int32_t syncEncodeSyncCfg(const void *pObj, SJson *pJson) {
if (nodeInfo == NULL) { if (nodeInfo == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY); TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
} }
if ((code = tjsonAddItemToObject(pJson, "nodeInfo", nodeInfo)) < 0) { if ((code = tjsonAddItemToObject(pJson, "nodeInfo", nodeInfo)) < 0) {
tjsonDelete(nodeInfo); tjsonDelete(nodeInfo);
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} }
for (int32_t i = 0; i < pCfg->totalReplicaNum; ++i) { for (int32_t i = 0; i < pCfg->totalReplicaNum; ++i) {
SJson *info = tjsonCreateObject(); SJson *info = tjsonCreateObject();
if (info == NULL) { if (info == NULL) {
@ -68,20 +69,25 @@ static int32_t syncEncodeSyncCfg(const void *pObj, SJson *pJson) {
TAOS_CHECK_GOTO(tjsonAddStringToObject(info, "isReplica", syncRoleToStr(pCfg->nodeInfo[i].nodeRole)), NULL, _err); TAOS_CHECK_GOTO(tjsonAddStringToObject(info, "isReplica", syncRoleToStr(pCfg->nodeInfo[i].nodeRole)), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddItemToArray(nodeInfo, info), NULL, _err); TAOS_CHECK_GOTO(tjsonAddItemToArray(nodeInfo, info), NULL, _err);
continue; continue;
_err: _err:
tjsonDelete(info); tjsonDelete(info);
break; break;
} }
_exit: _exit:
if (code < 0) { if (code < 0) {
sError("failed to encode sync cfg at line %d since %s", lino, tstrerror(code)); sError("failed to encode sync cfg at line %d since %s", lino, tstrerror(code));
} }
TAOS_RETURN(code); TAOS_RETURN(code);
} }
static int32_t syncEncodeRaftCfg(const void *pObj, SJson *pJson) { static int32_t syncEncodeRaftCfg(const void *pObj, SJson *pJson) {
SRaftCfg *pCfg = (SRaftCfg *)pObj; SRaftCfg *pCfg = (SRaftCfg *)pObj;
int32_t code = 0, lino = 0; int32_t code = 0;
int32_t lino = 0;
TAOS_CHECK_EXIT(tjsonAddObject(pJson, "SSyncCfg", syncEncodeSyncCfg, (void *)&pCfg->cfg)); TAOS_CHECK_EXIT(tjsonAddObject(pJson, "SSyncCfg", syncEncodeSyncCfg, (void *)&pCfg->cfg));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "isStandBy", pCfg->isStandBy)); TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "isStandBy", pCfg->isStandBy));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "snapshotStrategy", pCfg->snapshotStrategy)); TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "snapshotStrategy", pCfg->snapshotStrategy));
@ -93,10 +99,12 @@ static int32_t syncEncodeRaftCfg(const void *pObj, SJson *pJson) {
if (configIndexArr == NULL) { if (configIndexArr == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY); TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
} }
if ((code = tjsonAddItemToObject(pJson, "configIndexArr", configIndexArr)) < 0) { if ((code = tjsonAddItemToObject(pJson, "configIndexArr", configIndexArr)) < 0) {
tjsonDelete(configIndexArr); tjsonDelete(configIndexArr);
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} }
for (int32_t i = 0; i < pCfg->configIndexCount; ++i) { for (int32_t i = 0; i < pCfg->configIndexCount; ++i) {
SJson *configIndex = tjsonCreateObject(); SJson *configIndex = tjsonCreateObject();
if (configIndex == NULL) { if (configIndex == NULL) {
@ -105,14 +113,17 @@ static int32_t syncEncodeRaftCfg(const void *pObj, SJson *pJson) {
TAOS_CHECK_EXIT(tjsonAddIntegerToObject(configIndex, "index", pCfg->configIndexArr[i])); TAOS_CHECK_EXIT(tjsonAddIntegerToObject(configIndex, "index", pCfg->configIndexArr[i]));
TAOS_CHECK_EXIT(tjsonAddItemToArray(configIndexArr, configIndex)); TAOS_CHECK_EXIT(tjsonAddItemToArray(configIndexArr, configIndex));
continue; continue;
_err: _err:
tjsonDelete(configIndex); tjsonDelete(configIndex);
break; break;
} }
_exit: _exit:
if (code < 0) { if (code < 0) {
sError("failed to encode raft cfg at line %d since %s", lino, tstrerror(code)); sError("failed to encode raft cfg at line %d since %s", lino, tstrerror(code));
} }
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -124,11 +135,13 @@ int32_t syncWriteCfgFile(SSyncNode *pNode) {
const char *realfile = pNode->configPath; const char *realfile = pNode->configPath;
SRaftCfg *pCfg = &pNode->raftCfg; SRaftCfg *pCfg = &pNode->raftCfg;
char file[PATH_MAX] = {0}; char file[PATH_MAX] = {0};
(void)snprintf(file, sizeof(file), "%s.bak", realfile); (void)snprintf(file, sizeof(file), "%s.bak", realfile);
if ((pJson = tjsonCreateObject()) == NULL) { if ((pJson = tjsonCreateObject()) == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY); TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
} }
TAOS_CHECK_EXIT(tjsonAddObject(pJson, "RaftCfg", syncEncodeRaftCfg, pCfg)); TAOS_CHECK_EXIT(tjsonAddObject(pJson, "RaftCfg", syncEncodeRaftCfg, pCfg));
buffer = tjsonToString(pJson); buffer = tjsonToString(pJson);
if (buffer == NULL) { if (buffer == NULL) {
@ -145,6 +158,7 @@ int32_t syncWriteCfgFile(SSyncNode *pNode) {
if (taosWriteFile(pFile, buffer, len) <= 0) { if (taosWriteFile(pFile, buffer, len) <= 0) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno)); TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
} }
if (taosFsyncFile(pFile) < 0) { if (taosFsyncFile(pFile) < 0) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno)); TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
} }
@ -165,6 +179,7 @@ _exit:
if (code != 0) { if (code != 0) {
sError("vgId:%d, failed to write sync cfg file:%s since %s", pNode->vgId, realfile, tstrerror(code)); sError("vgId:%d, failed to write sync cfg file:%s since %s", pNode->vgId, realfile, tstrerror(code));
} }
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -232,6 +247,7 @@ static int32_t syncDecodeRaftCfg(const SJson *pJson, void *pObj) {
tjsonGetNumberValue(configIndex, "index", pCfg->configIndexArr[i], code); tjsonGetNumberValue(configIndex, "index", pCfg->configIndexArr[i], code);
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT; if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
} }
return 0; return 0;
} }
@ -292,16 +308,6 @@ _OVER:
if (code != 0) { if (code != 0) {
sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code)); sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code));
} }
TAOS_RETURN(code); TAOS_RETURN(code);
} }
int32_t syncAddCfgIndex(SSyncNode *pNode, SyncIndex cfgIndex) {
SRaftCfg *pCfg = &pNode->raftCfg;
if (pCfg->configIndexCount >= MAX_CONFIG_INDEX_COUNT) {
return TSDB_CODE_OUT_OF_RANGE;
}
pCfg->configIndexArr[pCfg->configIndexCount] = cfgIndex;
pCfg->configIndexCount++;
return 0;
}