fix: invalid read in multi-process mode

include/util/taoserror.h

@@ -87,6 +87,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MSG_NOT_PROCESSED             TAOS_DEF_ERROR_CODE(0, 0x0113)
 #define TSDB_CODE_CFG_NOT_FOUND                 TAOS_DEF_ERROR_CODE(0, 0x0114)
 #define TSDB_CODE_REPEAT_INIT                   TAOS_DEF_ERROR_CODE(0, 0x0115)
+#define TSDB_CODE_DUP_KEY                       TAOS_DEF_ERROR_CODE(0, 0x0116)
 
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0140)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0141)

source/dnode/mgmt/implement/src/dmTransport.c

@@ -72,6 +72,7 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
   NodeMsgFp msgFp = NULL;
   uint16_t  msgType = pRpc->msgType;
   bool      needRelease = false;
+  bool      isReq = msgType & 1U;
 
   if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) {
     dmSetMnodeEpSet(pWrapper->pDnode, pEpSet);
@@ -85,13 +86,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
   if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER;
 
   if (pWrapper->procType == DND_PROC_SINGLE) {
-    dTrace("msg:%p, is created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user);
+    dTrace("msg:%p, created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user);
     code = (*msgFp)(pWrapper, pMsg);
   } else if (pWrapper->procType == DND_PROC_PARENT) {
-    dTrace("msg:%p, is created and put into child queue, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType),
-           pRpc->handle, pMsg->user);
-    code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle,
-                               pRpc->refId, PROC_FUNC_REQ);
+    dTrace("msg:%p, created and put into child queue, type:%s handle:%p code:0x%04x user:%s contLen:%d", pMsg,
+           TMSG_INFO(msgType), pRpc->handle, pMsg->rpcMsg.code & 0XFFFF, pMsg->user, pRpc->contLen);
+    code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen,
+                               (isReq && (pMsg->rpcMsg.code == 0)) ? pRpc->handle : NULL, pRpc->refId, PROC_FUNC_REQ);
   } else {
     dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
     ASSERT(1);
@@ -100,12 +101,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
 _OVER:
   if (code == 0) {
     if (pWrapper->procType == DND_PROC_PARENT) {
-      dTrace("msg:%p, is freed in parent process", pMsg);
+      dTrace("msg:%p, freed in parent process", pMsg);
       taosFreeQitem(pMsg);
       rpcFreeCont(pRpc->pCont);
     }
   } else {
-    dError("msg:%p, type:%s failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), code & 0XFFFF, terrstr());
+    dError("msg:%p, type:%s handle:%p failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), pRpc->handle,
+           code & 0XFFFF, terrstr());
     if (msgType & 1U) {
       if (terrno != 0) code = terrno;
       if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) {
@@ -350,29 +352,32 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
 
 static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
                                  EProcFuncType ftype) {
+  int32_t code = pMsg->code & 0xFFFF;
   pMsg->pCont = pCont;
-  dTrace("msg:%p, get from parent queue, ftype:%d handle:%p code:0x%04x mtype:%d, app:%p", pMsg, ftype, pMsg->handle,
-         pMsg->code & 0xFFFF, pMsg->msgType, pMsg->ahandle);
-
-  switch (ftype) {
-    case PROC_FUNC_REGIST:
-      rpcRegisterBrokenLinkArg(pMsg);
-      break;
-    case PROC_FUNC_RELEASE:
-      taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
-      rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
-      rpcFreeCont(pCont);
-      break;
-    case PROC_FUNC_REQ:
-      dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
-      break;
-    case PROC_FUNC_RSP:
-      pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
-      dmSendRpcRsp(pWrapper->pDnode, pMsg);
-      break;
-    default:
-      break;
+
+  if (ftype == PROC_FUNC_REQ) {
+    dTrace("msg:%p, get from parent queue, send req:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType),
+           pMsg->handle, code, pMsg->ahandle);
+    dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
+  } else if (ftype == PROC_FUNC_RSP) {
+    dTrace("msg:%p, get from parent queue, send rsp:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType),
+           pMsg->handle, code, pMsg->ahandle);
+    pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
+    dmSendRpcRsp(pWrapper->pDnode, pMsg);
+  } else if (ftype == PROC_FUNC_REGIST) {
+    dTrace("msg:%p, get from parent queue, regist handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code,
+           pMsg->ahandle);
+    rpcRegisterBrokenLinkArg(pMsg);
+  } else if (ftype == PROC_FUNC_RELEASE) {
+    dTrace("msg:%p, get from parent queue, release handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code,
+           pMsg->ahandle);
+    taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
+    rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
+    rpcFreeCont(pCont);
+  } else {
+    dError("msg:%p, invalid ftype:%d while get from parent queue, handle:%p", pMsg, ftype, pMsg->handle);
   }
+
   taosMemoryFree(pMsg);
 }
 

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

@@ -136,7 +136,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
   // sync integration response
   for (int i = 0; i < taosArrayGetSize(pArray); i++) {
     SNodeMsg *pMsg;
-    SRpcMsg * pRpc;
+    SRpcMsg  *pRpc;
 
     pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     pRpc = &pMsg->rpcMsg;
@@ -175,7 +175,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg * pMsg = NULL;
+  SNodeMsg  *pMsg = NULL;
   SRpcMsg    rsp;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
@@ -218,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg * pMsg = NULL;
+  SNodeMsg  *pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -231,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
 
 static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg * pMsg = NULL;
+  SNodeMsg  *pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -248,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 }
 
 static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) {
-  SRpcMsg * pRpc = &pMsg->rpcMsg;
+  SRpcMsg  *pRpc = &pMsg->rpcMsg;
   SMsgHead *pHead = pRpc->pCont;
   pHead->contLen = ntohl(pHead->contLen);
   pHead->vgId = ntohl(pHead->vgId);
@@ -262,23 +262,23 @@ static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueTyp
   int32_t code = 0;
   switch (qtype) {
     case QUERY_QUEUE:
-      dTrace("msg:%p, will be written into vnode-query queue", pMsg);
+      dTrace("msg:%p, type:%s will be written into vnode-query queue", pMsg, TMSG_INFO(pRpc->msgType));
       taosWriteQitem(pVnode->pQueryQ, pMsg);
       break;
     case FETCH_QUEUE:
-      dTrace("msg:%p, will be written into vnode-fetch queue", pMsg);
+      dTrace("msg:%p, type:%s will be written into vnode-fetch queue", pMsg, TMSG_INFO(pRpc->msgType));
       taosWriteQitem(pVnode->pFetchQ, pMsg);
       break;
     case WRITE_QUEUE:
-      dTrace("msg:%p, will be written into vnode-write queue", pMsg);
+      dTrace("msg:%p, type:%s will be written into vnode-write queue", pMsg, TMSG_INFO(pRpc->msgType));
       taosWriteQitem(pVnode->pWriteQ, pMsg);
       break;
     case SYNC_QUEUE:
-      dTrace("msg:%p, will be written into vnode-sync queue", pMsg);
+      dTrace("msg:%p, type:%s will be written into vnode-sync queue", pMsg, TMSG_INFO(pRpc->msgType));
       taosWriteQitem(pVnode->pSyncQ, pMsg);
       break;
     case MERGE_QUEUE:
-      dTrace("msg:%p, will be written into vnode-merge queue", pMsg);
+      dTrace("msg:%p, type:%s will be written into vnode-merge queue", pMsg, TMSG_INFO(pRpc->msgType));
       taosWriteQitem(pVnode->pMergeQ, pMsg);
       break;
     default:
@@ -317,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->mgmtWorker;
   dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
   taosWriteQitem(pWorker->queue, pMsg);
@@ -325,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -335,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 
 static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
   SVnodesMgmt *pMgmt = pWrapper->pMgmt;
-  SMsgHead *   pHead = pRpc->pCont;
+  SMsgHead    *pHead = pRpc->pCont;
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;

source/util/src/terror.c

@@ -93,6 +93,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT,              "Operation not support
 TAOS_DEFINE_ERROR(TSDB_CODE_MSG_NOT_PROCESSED,            "Message not processed")
 TAOS_DEFINE_ERROR(TSDB_CODE_CFG_NOT_FOUND,                "Config not found")
 TAOS_DEFINE_ERROR(TSDB_CODE_REPEAT_INIT,                  "Repeat initialization")
+TAOS_DEFINE_ERROR(TSDB_CODE_DUP_KEY,                      "Cannot add duplicate keys to hash")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY,                "Ref out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL,                     "too many Ref Objs")

source/util/src/thash.c

@@ -310,6 +310,7 @@ int32_t taosHashGetSize(const SHashObj *pHashObj) {
 
 int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const void *data, size_t size) {
   if (pHashObj == NULL || key == NULL || keyLen == 0) {
+    terrno = TSDB_CODE_INVALID_PTR;
     return -1;
   }
 
@@ -378,6 +379,8 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const vo
       }
 
       doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode);
+    } else {
+      terrno = TSDB_CODE_DUP_KEY;
     }
 
     taosHashEntryWUnlock(pHashObj, pe);

source/util/src/tprocess.c

@@ -175,7 +175,6 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char
   if (handle != 0 && ftype == PROC_FUNC_REQ) {
     if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handleRef, sizeof(int64_t)) != 0) {
       taosThreadMutexUnlock(&pQueue->mutex);
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
     }
   }
@@ -227,8 +226,8 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char
   taosThreadMutexUnlock(&pQueue->mutex);
   tsem_post(&pQueue->sem);
 
-  uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d, head:%d %p body:%d %p", pQueue->name, pos, ftype,
-         pQueue->items, headLen, pHead, bodyLen, pBody);
+  uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d handle:%p ref:%" PRId64 ", head:%d %p body:%d %p",
+         pQueue->name, pos, ftype, pQueue->items, (void *)handle, handleRef, headLen, pHead, bodyLen, pBody);
   return 0;
 }