refact vnode write queue

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);

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.

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) {