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

fix(stream): fix race condition when starting check downstream status and dropping task are executed concurrently.

This commit is contained in:
Haojun Liao 2024-07-21 02:06:09 +08:00
parent 651077866e
commit 70191f6a5d
2 changed files with 10 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/libs/stream/src/streamCheckStatus.c
View File

@ -280,6 +280,13 @@ void streamTaskStartMonitorCheckRsp(SStreamTask* pTask) {
streamMutexLock(&pInfo->checkInfoLock); streamMutexLock(&pInfo->checkInfoLock);
// drop procedure already started, not start check downstream now
ETaskStatus s = streamTaskGetStatus(pTask).state;
if (s == TASK_STATUS__DROPPING) {
streamMutexUnlock(&pInfo->checkInfoLock);
return;
}
int32_t code = streamTaskStartCheckDownstream(pInfo, pTask->id.idStr); int32_t code = streamTaskStartCheckDownstream(pInfo, pTask->id.idStr);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
streamMutexUnlock(&pInfo->checkInfoLock); streamMutexUnlock(&pInfo->checkInfoLock);

3
source/libs/stream/src/streamMeta.c
View File

@ -742,7 +742,10 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
streamMetaRLock(pMeta); streamMetaRLock(pMeta);
ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id)); ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (ppTask) { if (ppTask) {
// to make sure check status will not start the check downstream status when we start to check timerActive count.
streamMutexLock(&pTask->taskCheckInfo.checkInfoLock);
timerActive = (*ppTask)->status.timerActive; timerActive = (*ppTask)->status.timerActive;
streamMutexUnlock(&pTask->taskCheckInfo.checkInfoLock);
} }
streamMetaRUnLock(pMeta); streamMetaRUnLock(pMeta);