shm

source/dnode/mgmt/snode/inc/smInt.h

@@ -49,6 +49,7 @@ void    smStopWorker(SSnodeMgmt *pMgmt);
 int32_t smProcessMgmtMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t smProcessUniqueMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t smProcessSharedMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg);
+int32_t smProcessExecMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg);
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/snode/src/smMsg.c

@@ -54,4 +54,8 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
   }
 }
 
-void smInitMsgHandles(SMgmtWrapper *pWrapper) {}
+void smInitMsgHandles(SMgmtWrapper *pWrapper) {
+  // Requests handled by SNODE
+  dndSetMsgHandle(pWrapper, TDMT_SND_TASK_DEPLOY, (NodeMsgFp)smProcessMgmtMsg);
+  dndSetMsgHandle(pWrapper, TDMT_SND_TASK_EXEC, (NodeMsgFp)smProcessExecMsg);
+}

source/dnode/mgmt/snode/src/smWorker.c

@@ -86,6 +86,12 @@ static FORCE_INLINE int32_t smGetSWIdFromMsg(SRpcMsg *pMsg) {
   return pHead->streamTaskId % SND_UNIQUE_THREAD_NUM;
 }
 
+static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
+  SStreamExecMsgHead *pHead = pMsg->pCont;
+  pHead->workerType = htonl(pHead->workerType);
+  return pHead->workerType;
+}
+
 int32_t smProcessMgmtMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
   SDnodeWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0);
   if (pWorker == NULL) {
@@ -115,3 +121,12 @@ int32_t smProcessSharedMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
   return dndWriteMsgToWorker(pWorker, pMsg);
 }
+
+int32_t smProcessExecMsg(SSnodeMgmt *pMgmt, SNodeMsg *pMsg) {
+  int32_t workerType = smGetSWTypeFromMsg(&pMsg->rpcMsg);
+  if (workerType == SND_WORKER_TYPE__SHARED) {
+    return smProcessSharedMsg(pMgmt, pMsg);
+  } else {
+    return smProcessUniqueMsg(pMgmt, pMsg);
+  }
+}