From 0addf0999693fcc1fcf3c5fc54e5c94049b0b942 Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Tue, 26 Jul 2022 11:24:39 +0800
Subject: [PATCH] refactor(sync): add pre-commit interface

---
 source/dnode/vnode/src/vnd/vnodeSync.c | 33 +++++++++++++++++++-------
 source/libs/sync/src/syncCommit.c      | 26 +++++---------------
 2 files changed, 31 insertions(+), 28 deletions(-)

diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c
index a0e2354f51..2b760efba0 100644
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
@@ -206,13 +206,13 @@ static void inline vnodeProposeBatchMsg(SVnode *pVnode, SRpcMsg **pMsgArr, bool
 }
 
 void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
-  SVnode  *pVnode = pInfo->ahandle;
-  int32_t  vgId = pVnode->config.vgId;
-  int32_t  code = 0;
-  SRpcMsg *pMsg = NULL;
-  int32_t  arrayPos = 0;
-  SRpcMsg **pMsgArr = taosMemoryCalloc(numOfMsgs, sizeof(SRpcMsg*));
-  bool    *pIsWeakArr = taosMemoryCalloc(numOfMsgs, sizeof(bool));
+  SVnode   *pVnode = pInfo->ahandle;
+  int32_t   vgId = pVnode->config.vgId;
+  int32_t   code = 0;
+  SRpcMsg  *pMsg = NULL;
+  int32_t   arrayPos = 0;
+  SRpcMsg **pMsgArr = taosMemoryCalloc(numOfMsgs, sizeof(SRpcMsg *));
+  bool     *pIsWeakArr = taosMemoryCalloc(numOfMsgs, sizeof(bool));
   vTrace("vgId:%d, get %d msgs from vnode-write queue", vgId, numOfMsgs);
 
   for (int32_t msg = 0; msg < numOfMsgs; msg++) {
@@ -506,7 +506,7 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
          syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
          syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
 
-  if (cbMeta.code == 0) {
+  if (cbMeta.code == 0 && cbMeta.isWeak == 0) {
     SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
     rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
     memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
@@ -529,6 +529,23 @@ static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMet
   vTrace("vgId:%d, pre-commit-cb is excuted, fsm:%p, index:%" PRId64 ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s",
          syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
          syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
+
+  if (cbMeta.code == 0 && cbMeta.isWeak == 1) {
+    SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
+    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+    memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
+    syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
+    rpcMsg.info.conn.applyIndex = cbMeta.index;
+    rpcMsg.info.conn.applyTerm = cbMeta.term;
+    tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
+  } else {
+    SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info};
+    vError("vgId:%d, sync pre-commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync),
+           pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code));
+    if (rsp.info.handle != NULL) {
+      tmsgSendRsp(&rsp);
+    }
+  }
 }
 
 static void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c
index b3cdd079a4..fd6577477f 100644
--- a/source/libs/sync/src/syncCommit.c
+++ b/source/libs/sync/src/syncCommit.c
@@ -67,11 +67,6 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
   for (SyncIndex index = syncNodeGetLastIndex(pSyncNode); index > pSyncNode->commitIndex; --index) {
     bool agree = syncAgree(pSyncNode, index);
 
-    if (gRaftDetailLog) {
-      sTrace("syncMaybeAdvanceCommitIndex syncAgree:%d, index:%" PRId64 ", pSyncNode->commitIndex:%" PRId64, agree,
-             index, pSyncNode->commitIndex);
-    }
-
     if (agree) {
       // term
       SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index);
@@ -82,20 +77,15 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
         // update commit index
         newCommitIndex = index;
 
-        if (gRaftDetailLog) {
-          sTrace("syncMaybeAdvanceCommitIndex maybe to update, newCommitIndex:%" PRId64
-                 " commit, pSyncNode->commitIndex:%" PRId64,
-                 newCommitIndex, pSyncNode->commitIndex);
-        }
-
         syncEntryDestory(pEntry);
         break;
       } else {
-        if (gRaftDetailLog) {
-          sTrace("syncMaybeAdvanceCommitIndex can not commit due to term not equal, pEntry->term:%" PRIu64
-                 ", pSyncNode->pRaftStore->currentTerm:%" PRIu64,
-                 pEntry->term, pSyncNode->pRaftStore->currentTerm);
-        }
+        do {
+          char logBuf[128];
+          snprintf(logBuf, sizeof(logBuf), "can not commit due to term not equal, index:%ld, term:%lu", pEntry->index,
+                   pEntry->term);
+          syncNodeEventLog(pSyncNode, logBuf);
+        } while (0);
       }
 
       syncEntryDestory(pEntry);
@@ -107,10 +97,6 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
     SyncIndex beginIndex = pSyncNode->commitIndex + 1;
     SyncIndex endIndex = newCommitIndex;
 
-    if (gRaftDetailLog) {
-      sTrace("syncMaybeAdvanceCommitIndex sync commit %" PRId64, newCommitIndex);
-    }
-
     // update commit index
     pSyncNode->commitIndex = newCommitIndex;
 
-- 
GitLab