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

fix(sync): batch propose

This commit is contained in:
Minghao Li 2022-07-06 10:32:17 +08:00
parent 103a9cd9e8
commit 8a4e251be7
2 changed files with 16 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/sync/sync.h
View File

@ -26,7 +26,7 @@ extern "C" {
extern bool gRaftDetailLog; extern bool gRaftDetailLog;
#define SYNC_MAX_BATCH_SIZE 100 #define SYNC_MAX_BATCH_SIZE 500
#define SYNC_INDEX_BEGIN 0 #define SYNC_INDEX_BEGIN 0
#define SYNC_INDEX_INVALID -1 #define SYNC_INDEX_INVALID -1
#define SYNC_TERM_INVALID 0xFFFFFFFFFFFFFFFF #define SYNC_TERM_INVALID 0xFFFFFFFFFFFFFFFF

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

@ -659,6 +659,18 @@ static bool syncNodeBatchOK(SRpcMsg* pMsgArr, int32_t arrSize) {
if (pMsgArr[i].msgType == TDMT_SYNC_CONFIG_CHANGE_FINISH) { if (pMsgArr[i].msgType == TDMT_SYNC_CONFIG_CHANGE_FINISH) {
return false; return false;
} }
if (pMsgArr[i].msgType == TDMT_SYNC_LEADER_TRANSFER) {
return false;
}
if (pMsgArr[i].msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
return false;
}
if (pMsgArr[i].msgType == TDMT_SYNC_SET_VNODE_STANDBY) {
return false;
}
} }
return true; return true;
@ -672,12 +684,12 @@ int32_t syncNodeProposeBatch(SSyncNode* pSyncNode, SRpcMsg* pMsgArr, bool* pIsWe
} }
if (arrSize > SYNC_MAX_BATCH_SIZE) { if (arrSize > SYNC_MAX_BATCH_SIZE) {
syncNodeErrorLog(pSyncNode, "sync propose match batch error"); syncNodeErrorLog(pSyncNode, "sync propose batch error");
terrno = TSDB_CODE_SYN_BATCH_ERROR; terrno = TSDB_CODE_SYN_BATCH_ERROR;
return -1; return -1;
} }
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
syncNodeErrorLog(pSyncNode, "sync propose not leader"); syncNodeErrorLog(pSyncNode, "sync propose not leader");
terrno = TSDB_CODE_SYN_NOT_LEADER; terrno = TSDB_CODE_SYN_NOT_LEADER;
return -1; return -1;
@ -711,7 +723,7 @@ int32_t syncNodeProposeBatch(SSyncNode* pSyncNode, SRpcMsg* pMsgArr, bool* pIsWe
// enqueue msg ok // enqueue msg ok
} else { } else {
sError("enqueue msg error, FpEqMsg is NULL"); sError("vgId:%d, enqueue msg error, FpEqMsg is NULL", pSyncNode->vgId);
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
} }