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

fix(stream): fix the deadlock.

This commit is contained in:
Haojun Liao 2024-09-20 14:22:59 +08:00
parent 3f5d6c4868
commit 8c15f59d68
2 changed files with 1 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3439,7 +3439,6 @@ int32_t streamStateGetFirst_rocksdb(SStreamState* pState, SWinKey* key) {
} }
int32_t streamStateGetGroupKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) { int32_t streamStateGetGroupKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
stDebug("streamStateGetGroupKVByCur_rocksdb");
if (!pCur) { if (!pCur) {
return -1; return -1;
} }
@ -3885,7 +3884,6 @@ SStreamStateCur* streamStateSessionSeekKeyPrev_rocksdb(SStreamState* pState, con
} }
int32_t streamStateSessionGetKVByCur_rocksdb(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen) { int32_t streamStateSessionGetKVByCur_rocksdb(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen) {
stDebug("streamStateSessionGetKVByCur_rocksdb");
if (!pCur) { if (!pCur) {
return -1; return -1;
} }
@ -3985,7 +3983,6 @@ SStreamStateCur* streamStateFillGetCur_rocksdb(SStreamState* pState, const SWinK
return NULL; return NULL;
} }
int32_t streamStateFillGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) { int32_t streamStateFillGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
stDebug("streamStateFillGetKVByCur_rocksdb");
if (!pCur) { if (!pCur) {
return -1; return -1;
} }

11
source/libs/stream/src/streamCheckpoint.c
View File

@ -900,9 +900,6 @@ static int32_t doChkptStatusCheck(SStreamTask* pTask) {
int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask); int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask);
stDebug("s-task:%s vgId:%d all checkpoint-trigger recv, quit from monitor checkpoint-trigger, ref:%d", id, vgId, stDebug("s-task:%s vgId:%d all checkpoint-trigger recv, quit from monitor checkpoint-trigger, ref:%d", id, vgId,
ref); ref);
// streamMutexUnlock(&pTask->lock);
// streamMetaReleaseTask(pTask->pMeta, pTask);
return -1; return -1;
} }
@ -911,9 +908,6 @@ static int32_t doChkptStatusCheck(SStreamTask* pTask) {
stWarn("s-task:%s vgId:%d checkpoint-trigger retrieve by previous checkpoint procedure, checkpointId:%" PRId64 stWarn("s-task:%s vgId:%d checkpoint-trigger retrieve by previous checkpoint procedure, checkpointId:%" PRId64
", quit, ref:%d", ", quit, ref:%d",
id, vgId, pTmrInfo->launchChkptId, ref); id, vgId, pTmrInfo->launchChkptId, ref);
// streamMutexUnlock(&pActiveInfo->lock);
// streamMetaReleaseTask(pTask->pMeta, pTask);
return -1; return -1;
} }
@ -922,9 +916,6 @@ static int32_t doChkptStatusCheck(SStreamTask* pTask) {
int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask); int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask);
stWarn("s-task:%s vgId:%d active checkpoint may be cleared, quit from retrieve checkpoint-trigger send tmr, ref:%d", stWarn("s-task:%s vgId:%d active checkpoint may be cleared, quit from retrieve checkpoint-trigger send tmr, ref:%d",
id, vgId, ref); id, vgId, ref);
// streamMutexUnlock(&pActiveInfo->lock);
// streamMetaReleaseTask(pTask->pMeta, pTask);
return -1; return -1;
} }
@ -1020,7 +1011,7 @@ void checkpointTriggerMonitorFn(void* param, void* tmrId) {
int32_t code = doChkptStatusCheck(pTask); int32_t code = doChkptStatusCheck(pTask);
if (code) { if (code) {
streamMutexUnlock(&pTask->lock); streamMutexUnlock(&pActiveInfo->lock);
streamMetaReleaseTask(pTask->pMeta, pTask); streamMetaReleaseTask(pTask->pMeta, pTask);
return; return;
} }