Merge pull request #10012 from taosdata/feature/mnode

adjust func paras

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -737,15 +737,9 @@ int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
   return 0;
 }
 
-static void dndProcessVnodeQueryQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) {
-  SRpcMsg *pRsp = NULL;
-  vnodeProcessQueryReq(pVnode->pImpl, pMsg, &pRsp);
-}
+static void dndProcessVnodeQueryQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) { vnodeProcessQueryMsg(pVnode->pImpl, pMsg); }
 
-static void dndProcessVnodeFetchQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) {
-  SRpcMsg *pRsp = NULL;
-  vnodeProcessFetchReq(pVnode->pImpl, pMsg, &pRsp);
-}
+static void dndProcessVnodeFetchQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) { vnodeProcessFetchMsg(pVnode->pImpl, pMsg); }
 
 static void dndProcessVnodeWriteQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOfMsgs) {
   SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));

source/dnode/vnode/inc/tq.h

@@ -318,8 +318,8 @@ int       tqRegisterContext(STqGroup*, void* ahandle);
 int       tqSendLaunchQuery(STqMsgItem*, int64_t offset);
 #endif
 
-int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp);
-int32_t tqProcessSetConnReq(STQ* pTq, char* msg, SRpcMsg** ppRsp);
+int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessSetConnReq(STQ* pTq, char* msg);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/inc/vnode.h

@@ -159,20 +159,18 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
  *
  * @param pVnode The vnode object.
  * @param pMsg The request message
- * @param pRsp The response message
  * @return int 0 for success, -1 for failure
  */
-int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
+int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
 
 /**
  * @brief Process a fetch message.
  *
  * @param pVnode The vnode object.
  * @param pMsg The request message
- * @param pRsp The response message
  * @return int 0 for success, -1 for failure
  */
-int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
+int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg);
 
 /* ------------------------ SVnodeCfg ------------------------ */
 /**

source/dnode/vnode/src/tq/tq.c

@@ -667,8 +667,8 @@ int tqItemSSize() {
 }
 #endif
 
-int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp) {
-  SMqConsumeReq* pReq = pMsg->pCont;
+int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
+  SMqConsumeReq*   pReq = pMsg->pCont;
   SRpcMsg          rpcMsg;
   int64_t          reqId = pReq->reqId;
   int64_t          consumerId = pReq->consumerId;
@@ -783,7 +783,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp) {
   return 0;
 }
 
-int32_t tqProcessSetConnReq(STQ* pTq, char* msg, SRpcMsg** ppRsp) {
+int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   SMqSetCVgReq req;
   tDecodeSMqSetCVgReq(msg, &req);
   STqConsumerHandle* pConsumer = calloc(sizeof(STqConsumerHandle), 1);

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

@@ -17,14 +17,14 @@
 #include "vnd.h"
 
 static int32_t vnodeGetTableList(SVnode *pVnode, SRpcMsg *pMsg);
-static int     vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
+static int     vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg);
 
 int vnodeQueryOpen(SVnode *pVnode) {
   return qWorkerInit(NODE_TYPE_VNODE, pVnode->vgId, NULL, (void **)&pVnode->pQuery, pVnode,
                      (putReqToQueryQFp)vnodePutReqToVQueryQ);
 }
 
-int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
+int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   vTrace("message in query queue is processing");
 
   switch (pMsg->msgType) {
@@ -38,7 +38,7 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   }
 }
 
-int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
+int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   vTrace("message in fetch queue is processing");
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
@@ -57,16 +57,16 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       return vnodeGetTableList(pVnode, pMsg);
       //      return qWorkerProcessShowFetchMsg(pVnode->pMeta, pVnode->pQuery, pMsg);
     case TDMT_VND_TABLE_META:
-      return vnodeGetTableMeta(pVnode, pMsg, pRsp);
+      return vnodeGetTableMeta(pVnode, pMsg);
     case TDMT_VND_CONSUME:
-      return tqProcessConsumeReq(pVnode->pTq, pMsg, pRsp);
+      return tqProcessConsumeReq(pVnode->pTq, pMsg);
     default:
       vError("unknown msg type:%d in fetch queue", pMsg->msgType);
       return TSDB_CODE_VND_APP_ERROR;
   }
 }
 
-static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
+static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   STableInfoReq * pReq = (STableInfoReq *)(pMsg->pCont);
   STbCfg *        pTbCfg = NULL;
   STbCfg *        pStbCfg = NULL;

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

@@ -115,7 +115,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       }
       break;
     case TDMT_VND_MQ_SET_CONN: {
-      if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead)), NULL) < 0) {
+      if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead))) < 0) {
         // TODO: handle error
       }
     } break;