From 50229512d32ad7c1690f12c2a976a7c9bbaba37f Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 21 Mar 2022 14:39:35 +0800
Subject: [PATCH] refact vnode write queue

---
 source/dnode/mgmt/vnode/src/vmWorker.c  | 32 ++++++++++++++++++-------
 source/dnode/vnode/inc/vnode.h          |  3 +--
 source/dnode/vnode/src/vnd/vnodeWrite.c |  5 ++--
 3 files changed, 27 insertions(+), 13 deletions(-)

diff --git a/source/dnode/mgmt/vnode/src/vmWorker.c b/source/dnode/mgmt/vnode/src/vmWorker.c
index 29a8678701..33945d438c 100644
--- a/source/dnode/mgmt/vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/vnode/src/vmWorker.c
@@ -63,36 +63,50 @@ static void vmProcessFetchQueue(SVnodeObj *pVnode, SNodeMsg *pMsg) {
   vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg);
 }
 
+static void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
+  SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
+  dndSendRsp(pWrapper, &rsp);
+}
+
 static void vmProcessWriteQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOfMsgs) {
   SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SNodeMsg *));
+  if (pArray == NULL) {
+    dError("failed to process %d msgs in write-queue since %s", numOfMsgs, terrstr());
+    return;
+  }
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     SNodeMsg *pMsg = NULL;
-    taosGetQitem(qall, (void **)&pMsg);
-    dTrace("msg:%p, will be processed in vnode write queue", pMsg);
-    void *ptr = taosArrayPush(pArray, &pMsg);
-    assert(ptr != NULL);
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+
+    dTrace("msg:%p, will be processed in vnode-write queue", pMsg);
+    if (taosArrayPush(pArray, &pMsg) == NULL) {
+      dTrace("msg:%p, failed to process since %s", pMsg, terrstr());
+      vmSendRsp(pVnode->pWrapper, pMsg, TSDB_CODE_OUT_OF_MEMORY);
+    }
   }
 
   vnodeProcessWMsgs(pVnode->pImpl, pArray);
 
-  for (size_t i = 0; i < numOfMsgs; i++) {
-    SRpcMsg  *pRsp = NULL;
+  numOfMsgs = taosArrayGetSize(pArray);
+  for (int32_t i = 0; i < numOfMsgs; i++) {
     SNodeMsg *pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     SRpcMsg  *pRpc = &pMsg->rpcMsg;
-    int32_t   code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp);
+    SRpcMsg  *pRsp = NULL;
+
+    int32_t code = vnodeApplyWMsg(pVnode->pImpl, pRpc, &pRsp);
     if (pRsp != NULL) {
       pRsp->ahandle = pRpc->ahandle;
       dndSendRsp(pVnode->pWrapper, pRsp);
       free(pRsp);
     } else {
-      if (code != 0) code = terrno;
+      if (code != 0 && terrno != 0) code = terrno;
       SRpcMsg rpcRsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
       dndSendRsp(pVnode->pWrapper, &rpcRsp);
     }
   }
 
-  for (size_t i = 0; i < numOfMsgs; i++) {
+  for (int32_t i = 0; i < numOfMsgs; i++) {
     SNodeMsg *pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     dTrace("msg:%p, is freed", pMsg);
     rpcFreeCont(pMsg->rpcMsg.pCont);
diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h
index d762c8e4c0..1ffd4e0d78 100644
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@@ -126,9 +126,8 @@ void vnodeDestroy(const char *path);
  *
  * @param pVnode The vnode object.
  * @param pMsgs The array of SRpcMsg
- * @return int 0 for success, -1 for failure
  */
-int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs);
+void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs);
 
 /**
  * @brief Apply a write request message.
diff --git a/source/dnode/vnode/src/vnd/vnodeWrite.c b/source/dnode/vnode/src/vnd/vnodeWrite.c
index 3fa987ab9b..d3769b8a30 100644
--- a/source/dnode/vnode/src/vnd/vnodeWrite.c
+++ b/source/dnode/vnode/src/vnd/vnodeWrite.c
@@ -16,7 +16,7 @@
 #include "tq.h"
 #include "vnd.h"
 
-int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
+void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
   SNodeMsg *pMsg;
   SRpcMsg *pRpc;
 
@@ -40,7 +40,8 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
 
   // TODO: Integrate RAFT module here
 
-  return 0;
+  // No results are returned because error handling is difficult
+  // return 0;
 }
 
 int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-- 
GitLab