Merge pull request #20927 from taosdata/FIX/TD-23552-main

enh: try to propose vnode commit at vnode closing

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
+#include "vnd.h"
 
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
@@ -78,6 +79,11 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
 
 void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal) {
   char path[TSDB_FILENAME_LEN] = {0};
+  bool atExit = true;
+
+  if (vnodeIsLeader(pVnode->pImpl)) {
+    vnodeProposeCommitOnNeed(pVnode->pImpl, atExit);
+  }
 
   taosThreadRwlockWrlock(&pMgmt->lock);
   taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t));

source/dnode/vnode/inc/vnode.h

@@ -92,7 +92,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
 void    vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
 void    vnodeApplyWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
-void    vnodeProposeCommitOnNeed(SVnode *pVnode);
+void    vnodeProposeCommitOnNeed(SVnode *pVnode, bool atExit);
 
 // meta
 typedef struct SMeta       SMeta;  // todo: remove

source/dnode/vnode/src/inc/vnd.h

@@ -96,7 +96,7 @@ int32_t vnodeGetBatchMeta(SVnode* pVnode, SRpcMsg* pMsg);
 
 // vnodeCommit.c
 int32_t vnodeBegin(SVnode* pVnode);
-int32_t vnodeShouldCommit(SVnode* pVnode);
+int32_t vnodeShouldCommit(SVnode* pVnode, bool atExit);
 void    vnodeUpdCommitSched(SVnode* pVnode);
 void    vnodeRollback(SVnode* pVnode);
 int32_t vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg);
@@ -115,7 +115,6 @@ void    vnodeSyncClose(SVnode* pVnode);
 void    vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg, int32_t code);
 bool    vnodeIsLeader(SVnode* pVnode);
 bool    vnodeIsRoleLeader(SVnode* pVnode);
-int     vnodeShouldCommit(SVnode* pVnode);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/src/vnd/vnodeCommit.c

@@ -149,7 +149,7 @@ void vnodeUpdCommitSched(SVnode *pVnode) {
   pVnode->commitSched.maxWaitMs = tsVndCommitMaxIntervalMs + (randNum % tsVndCommitMaxIntervalMs);
 }
 
-int vnodeShouldCommit(SVnode *pVnode) {
+int vnodeShouldCommit(SVnode *pVnode, bool atExit) {
   SVCommitSched *pSched = &pVnode->commitSched;
   int64_t        nowMs = taosGetMonoTimestampMs();
   bool           diskAvail = osDataSpaceAvailable();
@@ -158,7 +158,8 @@ int vnodeShouldCommit(SVnode *pVnode) {
   taosThreadMutexLock(&pVnode->mutex);
   if (pVnode->inUse && diskAvail) {
     needCommit =
-        ((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs));
+        ((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) ||
+        ((pVnode->inUse->size > 0) && atExit);
   }
   taosThreadMutexUnlock(&pVnode->mutex);
   return needCommit;

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -129,8 +129,8 @@ static int32_t inline vnodeProposeMsg(SVnode *pVnode, SRpcMsg *pMsg, bool isWeak
   return code;
 }
 
-void vnodeProposeCommitOnNeed(SVnode *pVnode) {
-  if (!vnodeShouldCommit(pVnode)) {
+void vnodeProposeCommitOnNeed(SVnode *pVnode, bool atExit) {
+  if (!vnodeShouldCommit(pVnode, atExit)) {
     return;
   }
 
@@ -145,18 +145,20 @@ void vnodeProposeCommitOnNeed(SVnode *pVnode) {
   rpcMsg.pCont = pHead;
   rpcMsg.info.noResp = 1;
 
+  vInfo("vgId:%d, propose vnode commit", pVnode->config.vgId);
   bool isWeak = false;
-  if (vnodeProposeMsg(pVnode, &rpcMsg, isWeak) < 0) {
-    vTrace("vgId:%d, failed to propose vnode commit since %s", pVnode->config.vgId, terrstr());
-    goto _out;
-  }
 
-  vInfo("vgId:%d, proposed vnode commit", pVnode->config.vgId);
+  if (!atExit) {
+    if (vnodeProposeMsg(pVnode, &rpcMsg, isWeak) < 0) {
+      vTrace("vgId:%d, failed to propose vnode commit since %s", pVnode->config.vgId, terrstr());
+    }
+    rpcFreeCont(rpcMsg.pCont);
+    rpcMsg.pCont = NULL;
+  } else {
+    tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &rpcMsg);
+  }
 
-_out:
   vnodeUpdCommitSched(pVnode);
-  rpcFreeCont(rpcMsg.pCont);
-  rpcMsg.pCont = NULL;
 }
 
 #if BATCH_ENABLE
@@ -236,7 +238,8 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
       continue;
     }
 
-    vnodeProposeCommitOnNeed(pVnode);
+    bool atExit = false;
+    vnodeProposeCommitOnNeed(pVnode, atExit);
 
     code = vnodePreProcessWriteMsg(pVnode, pMsg);
     if (code != 0) {
@@ -288,7 +291,8 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
       continue;
     }
 
-    vnodeProposeCommitOnNeed(pVnode);
+    bool atExit = false;
+    vnodeProposeCommitOnNeed(pVnode, atExit);
 
     code = vnodePreProcessWriteMsg(pVnode, pMsg);
     if (code != 0) {