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

fix(stream): concurrency dispatch

This commit is contained in:
Liu Jicong 2022-07-27 21:02:21 +08:00
parent ba0c1b3689
commit 904ec81bf9
2 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -328,6 +328,9 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
if (streamAddBlockToDispatchMsg(pDataBlock, &pReqs[j]) < 0) { if (streamAddBlockToDispatchMsg(pDataBlock, &pReqs[j]) < 0) {
goto FAIL_SHUFFLE_DISPATCH; goto FAIL_SHUFFLE_DISPATCH;
} }
if (pReqs[j].blockNum == 0) {
atomic_add_fetch_32(&pTask->shuffleDispatcher.waitingRspCnt, 1);
}
pReqs[j].blockNum++; pReqs[j].blockNum++;
found = true; found = true;
break; break;
@ -343,7 +346,6 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
if (streamDispatchOneReq(pTask, &pReqs[i], pVgInfo->vgId, &pVgInfo->epSet) < 0) { if (streamDispatchOneReq(pTask, &pReqs[i], pVgInfo->vgId, &pVgInfo->epSet) < 0) {
goto FAIL_SHUFFLE_DISPATCH; goto FAIL_SHUFFLE_DISPATCH;
} }
atomic_add_fetch_32(&pTask->shuffleDispatcher.waitingRspCnt, 1);
} }
} }
code = 0; code = 0;

5
tests/script/jenkins/basic.txt
View File

@ -298,8 +298,9 @@
# --- sma # --- sma
./test.sh -f tsim/sma/drop_sma.sim ./test.sh -f tsim/sma/drop_sma.sim
./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim ./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim # temp disable
./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim #./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
#./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
# --- valgrind # --- valgrind
./test.sh -f tsim/valgrind/checkError1.sim ./test.sh -f tsim/valgrind/checkError1.sim