refactor: rpc msg handler

source/dnode/mgmt/mgmt_qnode/src/qmWorker.c

@@ -19,70 +19,43 @@
 static inline void qmSendRsp(SRpcMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {
       .code = code,
-      .info = pMsg->info,
       .pCont = pMsg->info.rsp,
       .contLen = pMsg->info.rspLen,
+      .info = pMsg->info,
   };
   tmsgSendRsp(&rsp);
 }
 
-static void qmProcessMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
+static void qmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SQnodeMgmt *pMgmt = pInfo->ahandle;
+  int32_t     code = -1;
+  dTrace("msg:%p, get from qnode queue", pMsg);
 
-  dTrace("msg:%p, get from qnode-monitor queue", pMsg);
-  SRpcMsg *pRpc = pMsg;
-  int32_t  code = -1;
-
-  if (pMsg->msgType == TDMT_MON_QM_INFO) {
-    code = qmProcessGetMonitorInfoReq(pMgmt, pMsg);
-  } else {
-    terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+  switch (pMsg->msgType) {
+    case TDMT_MON_QM_INFO:
+      code = qmProcessGetMonitorInfoReq(pMgmt, pMsg);
+      break;
+    case TDMT_VND_QUERY:
+    case TDMT_VND_QUERY_CONTINUE:
+      code = qndProcessQueryMsg(pMgmt->pQnode, pMsg);
+      break;
+    default:
+      code = qndProcessFetchMsg(pMgmt->pQnode, pMsg);
+      break;
   }
 
-  if (pRpc->msgType & 1U) {
+  if (IsReq(pMsg) && code != TSDB_CODE_ACTION_IN_PROGRESS) {
     if (code != 0 && terrno != 0) code = terrno;
     qmSendRsp(pMsg, code);
   }
 
-  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
-  rpcFreeCont(pRpc->pCont);
-  taosFreeQitem(pMsg);
-}
-
-static void qmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
-  SQnodeMgmt *pMgmt = pInfo->ahandle;
-
-  dTrace("msg:%p, get from qnode-query queue", pMsg);
-  SRpcMsg *pRpc = pMsg;
-  int32_t  code = qndProcessQueryMsg(pMgmt->pQnode, pRpc);
-
-  if (pRpc->msgType & 1U && code != 0) {
-    qmSendRsp(pMsg, code);
-  }
-
-  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
-  rpcFreeCont(pMsg->pCont);
-  taosFreeQitem(pMsg);
-}
-
-static void qmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
-  SQnodeMgmt *pMgmt = pInfo->ahandle;
-
-  dTrace("msg:%p, get from qnode-fetch queue", pMsg);
-  SRpcMsg *pRpc = pMsg;
-  int32_t  code = qndProcessFetchMsg(pMgmt->pQnode, pRpc);
-
-  if (pRpc->msgType & 1U && code != 0) {
-    qmSendRsp(pMsg, code);
-  }
-
   dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
   rpcFreeCont(pMsg->pCont);
   taosFreeQitem(pMsg);
 }
 
 static int32_t qmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) {
-  dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
+  dTrace("msg:%p, put into worker %s, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType));
   taosWriteQitem(pWorker->queue, pMsg);
   return 0;
 }
@@ -101,9 +74,7 @@ int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
 static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) {
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
-  if (pMsg == NULL) {
-    return -1;
-  }
+  if (pMsg == NULL) return -1;
 
   dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType));
   memcpy(pMsg, pRpc, sizeof(SRpcMsg));
@@ -141,7 +112,7 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
       .min = tsNumOfVnodeQueryThreads,
       .max = tsNumOfVnodeQueryThreads,
       .name = "qnode-query",
-      .fp = (FItem)qmProcessQueryQueue,
+      .fp = (FItem)qmProcessQueue,
       .param = pMgmt,
   };
 
@@ -154,7 +125,7 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
       .min = tsNumOfQnodeFetchThreads,
       .max = tsNumOfQnodeFetchThreads,
       .name = "qnode-fetch",
-      .fp = (FItem)qmProcessFetchQueue,
+      .fp = (FItem)qmProcessQueue,
       .param = pMgmt,
   };
 
@@ -167,7 +138,7 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
       .min = 1,
       .max = 1,
       .name = "qnode-monitor",
-      .fp = (FItem)qmProcessMonitorQueue,
+      .fp = (FItem)qmProcessQueue,
       .param = pMgmt,
   };
   if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {

source/dnode/qnode/src/qnode.c

@@ -43,44 +43,63 @@ void qndClose(SQnode *pQnode) {
 int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) { return 0; }
 
 int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
-  qTrace("message in qnode query queue is processing");
+  int32_t     code = -1;
   SReadHandle handle = {.pMsgCb = &pQnode->msgCb};
+  qTrace("message in qnode query queue is processing");
 
   switch (pMsg->msgType) {
-    case TDMT_VND_QUERY: {
-      return qWorkerProcessQueryMsg(&handle, pQnode->pQuery, pMsg);
-    }
+    case TDMT_VND_QUERY:
+      code = qWorkerProcessQueryMsg(&handle, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_QUERY_CONTINUE:
-      return qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg);
+      code = qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg);
+      break;
     default:
       qError("unknown msg type:%d in query queue", pMsg->msgType);
-      return TSDB_CODE_VND_APP_ERROR;
+      terrno = TSDB_CODE_VND_APP_ERROR;
   }
+
+  if (code == 0) return TSDB_CODE_ACTION_IN_PROGRESS;
+  return code;
 }
 
 int32_t qndProcessFetchMsg(SQnode *pQnode, SRpcMsg *pMsg) {
+  int32_t code = -1;
   qTrace("message in fetch queue is processing");
+
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
-      return qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_FETCH_RSP:
-      return qWorkerProcessFetchRsp(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessFetchRsp(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_RES_READY:
-      return qWorkerProcessReadyMsg(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessReadyMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_TASKS_STATUS:
-      return qWorkerProcessStatusMsg(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessStatusMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_CANCEL_TASK:
-      return qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_DROP_TASK:
-      return qWorkerProcessDropMsg(pQnode, pQnode->pQuery, pMsg);
+      code = qWorkerProcessDropMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     case TDMT_VND_TABLE_META:
-      // return vnodeGetTableMeta(pQnode, pMsg);
+      // code =  vnodeGetTableMeta(pQnode, pMsg);
+      // break;
     case TDMT_VND_CONSUME:
-      // return tqProcessConsumeReq(pQnode->pTq, pMsg);
+      // code =  tqProcessConsumeReq(pQnode->pTq, pMsg);
+      // break;
     case TDMT_VND_QUERY_HEARTBEAT:
-      return qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg);      
+      code = qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg);
+      break;
     default:
       qError("unknown msg type:%d in fetch queue", pMsg->msgType);
-      return TSDB_CODE_VND_APP_ERROR;
+      terrno = TSDB_CODE_VND_APP_ERROR;
   }
+
+  if (code == 0) return TSDB_CODE_ACTION_IN_PROGRESS;
+  return code;
 }