xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

add stream exec head

This commit is contained in:
Liu Jicong 2022-03-18 16:26:32 +08:00
parent 552db887eb
commit d4d1c0a9f8
5 changed files with 50 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
include/common/tmsg.h
View File

@ -197,6 +197,11 @@ typedef struct {
};
} SMsgHead;
typedef struct {
int32_t workerType;
int32_t streamTaskId;
} SStreamExecMsgHead;
// Submit message for one table
typedef struct SSubmitBlk {
int64_t uid; // table unique id
@ -2310,7 +2315,7 @@ typedef struct {
} SStreamTaskDeployRsp;
typedef struct {
SMsgHead head;
SStreamExecMsgHead head;
// TODO: other info needed by task
} SStreamTaskExecReq;

5
include/util/tdef.h
View File

@ -448,6 +448,11 @@ typedef struct {
#define SND_UNIQUE_THREAD_NUM 2
#define SND_SHARED_THREAD_NUM 2
enum {
SND_WORKER_TYPE__SHARED = 1,
SND_WORKER_TYPE__UNIQUE,
};
#ifdef __cplusplus
}
#endif

7
source/dnode/mgmt/impl/inc/dndSnode.h
View File

@ -24,10 +24,15 @@ extern "C" {
int32_t dndInitSnode(SDnode *pDnode);
void dndCleanupSnode(SDnode *pDnode);
void dndProcessSnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
// void dndProcessSnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessSnodeSharedMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessSnodeExecMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
#ifdef __cplusplus
}
#endif

18
source/dnode/mgmt/impl/src/dndSnode.c
View File

@ -382,6 +382,12 @@ static void dndProcessSnodeSharedQueue(SDnode *pDnode, SRpcMsg *pMsg) {
taosFreeQitem(pMsg);
}
static FORCE_INLINE int32_t dndGetSWTypeFromMsg(SRpcMsg *pMsg) {
SStreamExecMsgHead *pHead = pMsg->pCont;
pHead->workerType = htonl(pHead->workerType);
return pHead->workerType;
}
static FORCE_INLINE int32_t dndGetSWIdFromMsg(SRpcMsg *pMsg) {
SMsgHead *pHead = pMsg->pCont;
pHead->streamTaskId = htonl(pHead->streamTaskId);
@ -450,6 +456,18 @@ void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndWriteSnodeMsgToMgmtWorker(pDnode, pMsg);
}
void dndProcessSnodeExecMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SSnode *pSnode = dndAcquireSnode(pDnode);
if (pSnode != NULL) {
int32_t workerType = dndGetSWTypeFromMsg(pMsg);
if (workerType == SND_WORKER_TYPE__SHARED) {
dndWriteSnodeMsgToWorker(pDnode, &pDnode->smgmt.sharedWorker, pMsg);
} else {
dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
}
}
}
void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
}

5
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -23,6 +23,7 @@
#include "dndTransport.h"
#include "dndMgmt.h"
#include "dndMnode.h"
#include "dndSnode.h"
#include "dndVnodes.h"
#define INTERNAL_USER "_dnd"
@ -153,6 +154,10 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CONSUME)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_HEARTBEAT)] = dndProcessVnodeFetchMsg;
// Requests handled by SNODE
pMgmt->msgFp[TMSG_INDEX(TDMT_SND_TASK_DEPLOY)] = dndProcessSnodeMgmtMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_SND_TASK_EXEC)] = dndProcessSnodeExecMsg;
}
static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {