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

enh: adjust sync propose

This commit is contained in:
Shengliang Guan 2022-11-07 21:12:43 +08:00
parent 4ddd25a29c
commit 848fd58402
3 changed files with 44 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -656,38 +656,34 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
}
int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
int32_t ret = 0;
sNTrace(pSyncNode, "propose message, type:%s", TMSG_INFO(pMsg->msgType));
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
// not restored, vnode enable
if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) {
ret = -1;
terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
sError("vgId:%d, failed to sync propose since not ready, type:%s, last:%" PRId64 ", cmt:%" PRId64,
pSyncNode->vgId, TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
goto _END;
if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
terrno = TSDB_CODE_SYN_NOT_LEADER;
sNError(pSyncNode, "sync propose not leader, %s, type:%s", syncStr(pSyncNode->state), TMSG_INFO(pMsg->msgType));
return -1;
}
SRespStub stub = {.createTime = taosGetTimestampMs(), .rpcMsg = *pMsg};
uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
// not restored, vnode enable
if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) {
terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
sNError(pSyncNode, "failed to sync propose since not ready, type:%s, last:%" PRId64 ", cmt:%" PRId64,
TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
return -1;
}
SyncClientRequest* pSyncMsg = syncClientRequestBuild(pMsg, seqNum, isWeak, pSyncNode->vgId);
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
int32_t ret = 0;
SyncClientRequest* pSyncMsg;
// optimized one replica
if (syncNodeIsOptimizedOneReplica(pSyncNode, pMsg)) {
pSyncMsg = syncClientRequestBuild(pMsg, 0, isWeak, pSyncNode->vgId);
SyncIndex retIndex;
int32_t code = syncNodeOnClientRequest(pSyncNode, pSyncMsg, &retIndex);
if (code == 0) {
pMsg->info.conn.applyIndex = retIndex;
pMsg->info.conn.applyTerm = pSyncNode->pRaftStore->currentTerm;
rpcFreeCont(rpcMsg.pCont);
syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
ret = 1;
sDebug("vgId:%d, sync optimize index:%" PRId64 ", type:%s", pSyncNode->vgId, retIndex,
TMSG_INFO(pMsg->msgType));
sTrace("vgId:%d, sync optimize index:%" PRId64 ", type:%s", pSyncNode->vgId, retIndex, TMSG_INFO(pMsg->msgType));
} else {
ret = -1;
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
@ -695,25 +691,22 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
TMSG_INFO(pMsg->msgType));
}
} else {
SRespStub stub = {.createTime = taosGetTimestampMs(), .rpcMsg = *pMsg};
uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
pSyncMsg = syncClientRequestBuild(pMsg, seqNum, isWeak, pSyncNode->vgId);
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
sNTrace(pSyncNode, "propose message, type:%s", TMSG_INFO(pMsg->msgType));
ret = (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg);
if (ret != 0) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
sError("vgId:%d, failed to enqueue msg since its null", pSyncNode->vgId);
sError("vgId:%d, failed to enqueue msg since %s", pSyncNode->vgId, terrstr());
}
}
syncClientRequestDestroy(pSyncMsg);
goto _END;
} else {
ret = -1;
terrno = TSDB_CODE_SYN_NOT_LEADER;
sError("vgId:%d, sync propose not leader, %s, type:%s", pSyncNode->vgId, syncStr(pSyncNode->state),
TMSG_INFO(pMsg->msgType));
goto _END;
}
_END:
return ret;
}
@ -2522,7 +2515,7 @@ int32_t syncNodeOnClientRequest(SSyncNode* ths, SyncClientRequest* pMsg, SyncInd
SyncIndex index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
SyncTerm term = ths->pRaftStore->currentTerm;
SSyncRaftEntry* pEntry = syncEntryBuild2((SyncClientRequest*)pMsg, term, index);
SSyncRaftEntry* pEntry = syncEntryBuild2(pMsg, term, index);
ASSERT(pEntry != NULL);
LRUHandle* h = NULL;

2
source/libs/sync/src/syncMessage.c
View File

@ -836,8 +836,6 @@ SyncClientRequest* syncClientRequestAlloc(uint32_t dataLen) {
SyncClientRequest* pMsg = taosMemoryCalloc(1, bytes);
pMsg->bytes = bytes;
pMsg->msgType = TDMT_SYNC_CLIENT_REQUEST;
pMsg->seqNum = 0;
pMsg->isWeak = false;
pMsg->dataLen = dataLen;
return pMsg;
}

24
source/libs/sync/src/syncUtil.c
View File

@ -178,29 +178,11 @@ void syncUtilMsgNtoH(void* msg) {
pHead->vgId = ntohl(pHead->vgId);
}
bool syncUtilUserPreCommit(tmsg_t msgType) {
if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER) {
return true;
}
bool syncUtilUserPreCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; }
return false;
}
bool syncUtilUserCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; }
bool syncUtilUserCommit(tmsg_t msgType) {
if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER) {
return true;
}
return false;
}
bool syncUtilUserRollback(tmsg_t msgType) {
if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER) {
return true;
}
return false;
}
bool syncUtilUserRollback(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; }
void syncCfg2SimpleStr(const SSyncCfg* pCfg, char* buf, int32_t bufLen) {
int32_t len = snprintf(buf, bufLen, "{r-num:%d, my:%d, ", pCfg->replicaNum, pCfg->myIndex);