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

fix(stream): check status before start timer.

This commit is contained in:
Haojun Liao 2024-08-15 16:56:30 +08:00
parent 644f97f1c1
commit 59270dfd0d
1 changed files with 19 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -762,18 +762,26 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) {
code = sendDispatchMsg(pTask, pTask->msgInfo.pData);
streamMutexLock(&pTask->msgInfo.lock);
if (pTask->msgInfo.inMonitor == 0) {
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s start dispatch monitor tmr in %dms, ref:%d, dispatch code:%s", id, DISPATCH_RETRY_INTERVAL_MS,
ref, tstrerror(code));
streamStartMonitorDispatchData(pTask, DISPATCH_RETRY_INTERVAL_MS);
pTask->msgInfo.inMonitor = 1;
} else {
stDebug("s-task:%s already in dispatch monitor tmr", id);
}
streamMutexLock(&pTask->lock);
bool shouldStop = streamTaskShouldStop(pTask);
streamMutexLock(&pTask->lock);
streamMutexUnlock(&pTask->msgInfo.lock);
if (shouldStop) {
stDebug("s-task:%s in stop/dropping status, not start dispatch monitor tmr", id);
} else {
streamMutexLock(&pTask->msgInfo.lock);
if (pTask->msgInfo.inMonitor == 0) {
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s start dispatch monitor tmr in %dms, ref:%d, dispatch code:%s", id, DISPATCH_RETRY_INTERVAL_MS,
ref, tstrerror(code));
streamStartMonitorDispatchData(pTask, DISPATCH_RETRY_INTERVAL_MS);
pTask->msgInfo.inMonitor = 1;
} else {
stDebug("s-task:%s already in dispatch monitor tmr", id);
}
streamMutexUnlock(&pTask->msgInfo.lock);
}
// this block can not be deleted until it has been sent to downstream task successfully.
return TSDB_CODE_SUCCESS;