add stream msg routing

include/common/tmsg.h

@@ -187,7 +187,10 @@ typedef struct SEp {
 
 typedef struct {
   int32_t contLen;
+  union {
     int32_t vgId;
+    int32_t streamTaskId;
+  };
 } SMsgHead;
 
 // Submit message for one table

include/dnode/snode/snode.h

@@ -23,6 +23,9 @@
 extern "C" {
 #endif
 
+#define SND_UNIQUE_THREAD_NUM 2
+#define SND_SHARED_THREAD_NUM 2
+
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SDnode SDnode;
 typedef struct SSnode SSnode;

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

@@ -166,7 +166,7 @@ static int32_t dndWriteSnodeFile(SDnode *pDnode) {
 static int32_t dndStartSnodeWorker(SDnode *pDnode) {
   SSnodeMgmt *pMgmt = &pDnode->smgmt;
   pMgmt->uniqueWorkers = taosArrayInit(0, sizeof(void *));
-  for (int32_t i = 0; i < 2; i++) {
+  for (int32_t i = 0; i < SND_UNIQUE_THREAD_NUM; i++) {
     SDnodeWorker *pUniqueWorker = malloc(sizeof(SDnodeWorker));
     if (pUniqueWorker == NULL) {
       return -1;
@@ -177,8 +177,8 @@ static int32_t dndStartSnodeWorker(SDnode *pDnode) {
     }
     taosArrayPush(pMgmt->uniqueWorkers, &pUniqueWorker);
   }
-  if (dndInitWorker(pDnode, &pMgmt->sharedWorker, DND_WORKER_SINGLE, "snode-shared", 4, 4,
-                    dndProcessSnodeSharedQueue)) {
+  if (dndInitWorker(pDnode, &pMgmt->sharedWorker, DND_WORKER_SINGLE, "snode-shared", SND_SHARED_THREAD_NUM,
+                    SND_SHARED_THREAD_NUM, dndProcessSnodeSharedQueue)) {
     dError("failed to start snode shared worker since %s", terrstr());
     return -1;
   }
@@ -369,13 +369,39 @@ static void dndProcessSnodeSharedQueue(SDnode *pDnode, SRpcMsg *pMsg) {
   taosFreeQitem(pMsg);
 }
 
-static void dndWriteSnodeMsgToRandomWorker(SDnode *pDnode, SRpcMsg *pMsg) {
+static FORCE_INLINE int32_t dndGetSWIdFromMsg(SRpcMsg *pMsg) {
+  SMsgHead *pHead = pMsg->pCont;
+  pHead->streamTaskId = htonl(pHead->streamTaskId);
+  return pHead->streamTaskId % SND_UNIQUE_THREAD_NUM;
+}
+
+static void dndWriteSnodeMsgToWorkerByMsg(SDnode *pDnode, SRpcMsg *pMsg) {
   int32_t code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
 
   SSnode *pSnode = dndAcquireSnode(pDnode);
   if (pSnode != NULL) {
-    int32_t       index = (pDnode->smgmt.uniqueWorkerInUse + 1) % taosArrayGetSize(pDnode->smgmt.uniqueWorkers);
-    SDnodeWorker *pWorker = taosArrayGet(pDnode->smgmt.uniqueWorkers, index);
+    int32_t       index = dndGetSWIdFromMsg(pMsg);
+    SDnodeWorker *pWorker = taosArrayGetP(pDnode->smgmt.uniqueWorkers, index);
+    code = dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg));
+  }
+
+  dndReleaseSnode(pDnode, pSnode);
+
+  if (code != 0) {
+    if (pMsg->msgType & 1u) {
+      SRpcMsg rsp = {.handle = pMsg->handle, .ahandle = pMsg->ahandle, .code = code};
+      rpcSendResponse(&rsp);
+    }
+    rpcFreeCont(pMsg->pCont);
+  }
+}
+
+static void dndWriteSnodeMsgToMgmtWorker(SDnode *pDnode, SRpcMsg *pMsg) {
+  int32_t code = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
+
+  SSnode *pSnode = dndAcquireSnode(pDnode);
+  if (pSnode != NULL) {
+    SDnodeWorker *pWorker = taosArrayGet(pDnode->smgmt.uniqueWorkers, 0);
     code = dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg));
   }
   dndReleaseSnode(pDnode, pSnode);
@@ -407,9 +433,12 @@ static void dndWriteSnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpc
   }
 }
 
+void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  dndWriteSnodeMsgToMgmtWorker(pDnode, pMsg);
+}
+
 void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  // judge from msg to write to unique queue
-  dndWriteSnodeMsgToRandomWorker(pDnode, pMsg);
+  dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
 }
 
 void dndProcessSnodeSharedMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {

source/dnode/snode/inc/sndInt.h

@@ -30,20 +30,25 @@ extern "C" {
 #endif
 
 enum {
-  STREAM_STATUS__READY = 1,
+  STREAM_STATUS__RUNNING = 1,
   STREAM_STATUS__STOPPED,
   STREAM_STATUS__CREATING,
   STREAM_STATUS__STOPING,
-  STREAM_STATUS__RESUMING,
+  STREAM_STATUS__RESTORING,
   STREAM_STATUS__DELETING,
 };
 
 enum {
-  STREAM_RUNNER__RUNNING = 1,
-  STREAM_RUNNER__STOP,
+  STREAM_TASK_STATUS__RUNNING = 1,
+  STREAM_TASK_STATUS__STOP,
 };
 
+typedef struct {
+  SHashObj* pHash;  // taskId -> streamTask
+} SStreamMeta;
+
 typedef struct SSnode {
+  SStreamMeta* pMeta;
   SSnodeOpt    cfg;
 } SSnode;
 
@@ -62,10 +67,6 @@ typedef struct {
   // storage handle
 } SStreamRunner;
 
-typedef struct {
-  SHashObj* pHash;
-} SStreamMeta;
-
 int32_t sndCreateStream();
 int32_t sndDropStream();