From 90e4cab2af624ecb5e3680d4bf8b203c4157444e Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 28 Mar 2022 14:16:58 +0800 Subject: [PATCH] fix stream serialization --- include/common/tmsgdef.h | 1 - source/dnode/mgmt/vnode/src/vmMsg.c | 1 - source/dnode/mnode/impl/src/mndDef.c | 17 +++++++++-------- source/dnode/vnode/src/inc/vnd.h | 2 +- source/dnode/vnode/src/tq/tq.c | 6 ++---- source/dnode/vnode/src/vnd/vnodeQuery.c | 5 +++-- source/dnode/vnode/src/vnd/vnodeWrite.c | 5 +++++ source/libs/stream/src/tstream.c | 21 ++++++++++++++++++--- 8 files changed, 38 insertions(+), 20 deletions(-) diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 051ee34644..36a489eb59 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -192,7 +192,6 @@ enum { TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp) TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp) TD_DEF_MSG_TYPE(TDMT_VND_TASK_DEPLOY, "vnode-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp) - TD_DEF_MSG_TYPE(TDMT_VND_TASK_EXEC, "vnode-task-exec", SStreamTaskExecReq, SStreamTaskExecRsp) TD_DEF_MSG_TYPE(TDMT_VND_TASK_PIPE_EXEC, "vnode-task-pipe-exec", SStreamTaskExecReq, SStreamTaskExecRsp) TD_DEF_MSG_TYPE(TDMT_VND_TASK_MERGE_EXEC, "vnode-task-merge-exec", SStreamTaskExecReq, SStreamTaskExecRsp) TD_DEF_MSG_TYPE(TDMT_VND_TASK_WRITE_EXEC, "vnode-task-write-exec", SStreamTaskExecReq, SStreamTaskExecRsp) diff --git a/source/dnode/mgmt/vnode/src/vmMsg.c b/source/dnode/mgmt/vnode/src/vmMsg.c index 97d829571f..1682c6043d 100644 --- a/source/dnode/mgmt/vnode/src/vmMsg.c +++ b/source/dnode/mgmt/vnode/src/vmMsg.c @@ -279,7 +279,6 @@ void vmInitMsgHandles(SMgmtWrapper *pWrapper) { dndSetMsgHandle(pWrapper, TDMT_VND_CONSUME, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY, (NodeMsgFp)vmProcessWriteMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); - dndSetMsgHandle(pWrapper, TDMT_VND_TASK_EXEC, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_PIPE_EXEC, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_TASK_MERGE_EXEC, (NodeMsgFp)vmProcessMergeMsg, VND_VGID); dndSetMsgHandle(pWrapper, TDMT_VND_STREAM_TRIGGER, (NodeMsgFp)vmProcessFetchMsg, VND_VGID); diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c index 24f2a5df22..1b3564924a 100644 --- a/source/dnode/mnode/impl/src/mndDef.c +++ b/source/dnode/mnode/impl/src/mndDef.c @@ -36,11 +36,11 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) { if (tEncodeI32(pEncoder, sz) < 0) return -1; for (int32_t i = 0; i < sz; i++) { - SArray *pArray = taosArrayGet(pObj->tasks, i); + SArray *pArray = taosArrayGetP(pObj->tasks, i); int32_t innerSz = taosArrayGetSize(pArray); if (tEncodeI32(pEncoder, innerSz) < 0) return -1; for (int32_t j = 0; j < innerSz; j++) { - SStreamTask *pTask = taosArrayGet(pArray, j); + SStreamTask *pTask = taosArrayGetP(pArray, j); if (tEncodeSStreamTask(pEncoder, pTask) < 0) return -1; } } @@ -76,17 +76,18 @@ int32_t tDecodeSStreamObj(SCoder *pDecoder, SStreamObj *pObj) { int32_t sz; if (tDecodeI32(pDecoder, &sz) < 0) return -1; if (sz != 0) { - pObj->tasks = taosArrayInit(sz, sizeof(SArray)); + pObj->tasks = taosArrayInit(sz, sizeof(void *)); for (int32_t i = 0; i < sz; i++) { int32_t innerSz; if (tDecodeI32(pDecoder, &innerSz) < 0) return -1; - SArray *pArray = taosArrayInit(innerSz, sizeof(SStreamTask)); + SArray *pArray = taosArrayInit(innerSz, sizeof(void *)); for (int32_t j = 0; j < innerSz; j++) { - SStreamTask task; - if (tDecodeSStreamTask(pDecoder, &task) < 0) return -1; - taosArrayPush(pArray, &task); + SStreamTask *pTask = taosMemoryCalloc(1, sizeof(SStreamTask)); + if (pTask == NULL) return -1; + if (tDecodeSStreamTask(pDecoder, pTask) < 0) return -1; + taosArrayPush(pArray, &pTask); } - taosArrayPush(pObj->tasks, pArray); + taosArrayPush(pObj->tasks, &pArray); } } diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index 8d256995c6..3309686bb7 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -198,7 +198,7 @@ int tqCommit(STQ*); int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessSetConnReq(STQ* pTq, char* msg); int32_t tqProcessRebReq(STQ* pTq, char* msg); -int32_t tqProcessTaskExec(STQ* pTq, SRpcMsg* msg); +int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 4661668cbe..c354b01501 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -497,11 +497,9 @@ int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen) { return 0; } -int32_t tqProcessTaskExec(STQ* pTq, SRpcMsg* msg) { - char* msgstr = POINTER_SHIFT(msg->pCont, sizeof(SMsgHead)); - +int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen) { SStreamTaskExecReq req; - tDecodeSStreamTaskExecReq(msgstr, &req); + tDecodeSStreamTaskExecReq(msg, &req); int32_t taskId = req.taskId; SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index 94e183f525..74d7558e0d 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -43,6 +43,8 @@ int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) { vTrace("message in fetch queue is processing"); + char *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); + int32_t msgLen = pMsg->contLen - sizeof(SMsgHead); switch (pMsg->msgType) { case TDMT_VND_FETCH: return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg); @@ -65,10 +67,9 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) { return vnodeGetTableMeta(pVnode, pMsg); case TDMT_VND_CONSUME: return tqProcessPollReq(pVnode->pTq, pMsg); - case TDMT_VND_TASK_EXEC: case TDMT_VND_TASK_PIPE_EXEC: case TDMT_VND_TASK_MERGE_EXEC: - return tqProcessTaskExec(pVnode->pTq, pMsg); + return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen); case TDMT_VND_STREAM_TRIGGER: return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen); case TDMT_VND_QUERY_HEARTBEAT: diff --git a/source/dnode/vnode/src/vnd/vnodeWrite.c b/source/dnode/vnode/src/vnd/vnodeWrite.c index a14828cc22..279926747f 100644 --- a/source/dnode/vnode/src/vnd/vnodeWrite.c +++ b/source/dnode/vnode/src/vnd/vnodeWrite.c @@ -167,6 +167,11 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { pMsg->contLen - sizeof(SMsgHead)) < 0) { } } break; + case TDMT_VND_TASK_WRITE_EXEC: { + if (tqProcessTaskExec(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), + pMsg->contLen - sizeof(SMsgHead)) < 0) { + } + } break; case TDMT_VND_CREATE_SMA: { // timeRangeSMA #if 0 SSmaCfg vCreateSmaReq = {0}; diff --git a/source/libs/stream/src/tstream.c b/source/libs/stream/src/tstream.c index 31a06a9d9a..6d69a8960e 100644 --- a/source/libs/stream/src/tstream.c +++ b/source/libs/stream/src/tstream.c @@ -205,9 +205,16 @@ int32_t tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask) { if (tEncodeCStr(pEncoder, pTask->exec.qmsg) < 0) return -1; } - if (pTask->sinkType != TASK_SINK__NONE) { - // TODO: wrap + if (pTask->sinkType == TASK_SINK__TABLE) { if (tEncodeI8(pEncoder, pTask->tbSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__SMA) { + if (tEncodeI8(pEncoder, pTask->smaSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__FETCH) { + if (tEncodeI8(pEncoder, pTask->fetchSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__SHOW) { + if (tEncodeI8(pEncoder, pTask->showSink.reserved) < 0) return -1; + } else { + ASSERT(pTask->sinkType == TASK_SINK__NONE); } if (pTask->dispatchType == TASK_DISPATCH__INPLACE) { @@ -244,8 +251,16 @@ int32_t tDecodeSStreamTask(SCoder* pDecoder, SStreamTask* pTask) { if (tDecodeCStrAlloc(pDecoder, &pTask->exec.qmsg) < 0) return -1; } - if (pTask->sinkType != TASK_SINK__NONE) { + if (pTask->sinkType == TASK_SINK__TABLE) { if (tDecodeI8(pDecoder, &pTask->tbSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__SMA) { + if (tDecodeI8(pDecoder, &pTask->smaSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__FETCH) { + if (tDecodeI8(pDecoder, &pTask->fetchSink.reserved) < 0) return -1; + } else if (pTask->sinkType == TASK_SINK__SHOW) { + if (tDecodeI8(pDecoder, &pTask->showSink.reserved) < 0) return -1; + } else { + ASSERT(pTask->sinkType == TASK_SINK__NONE); } if (pTask->dispatchType == TASK_DISPATCH__INPLACE) {