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

fix(stream): dispatch

This commit is contained in:
Liu Jicong 2022-07-27 11:44:01 +08:00
parent 3fa5791889
commit 57f9d8b0b0
2 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -118,12 +118,12 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, SStreamObj* pStream, SStream
bool isShuffle = false;
if (pStream->fixedSinkVgId == 0) {
pTask->dispatchType = TASK_DISPATCH__SHUFFLE;
pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
ASSERT(pDb);
if (pDb->cfg.numOfVgroups > 1) {
isShuffle = true;
pTask->dispatchType = TASK_DISPATCH__SHUFFLE;
pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
ASSERT(0);
return -1;

6
source/libs/stream/src/streamDispatch.c
View File

@ -280,7 +280,7 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
taosArrayDestroy(req.dataLen);
return code;
} else if (pTask->dispatchMsgType == TASK_DISPATCH__SHUFFLE) {
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
ASSERT(pTask->shuffleDispatcher.waitingRspCnt == 0);
int32_t vgSz = taosArrayGetSize(vgInfo);
@ -318,6 +318,8 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
if (streamAddBlockToDispatchMsg(pDataBlock, &pReqs[j]) < 0) {
goto FAIL_SHUFFLE_DISPATCH;
}
pReqs[j].taskId = pVgInfo->taskId;
pReqs[j].blockNum++;
break;
}
}
@ -342,6 +344,8 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
taosMemoryFree(pReqs);
}
return code;
} else {
ASSERT(0);
}
return 0;
}