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

fix stream backend crash when transfer

This commit is contained in:
yihaoDeng 2023-12-18 11:19:15 +08:00
parent 5aa575acfa
commit 0840e6a0d8
1 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -260,9 +260,17 @@ int32_t streamTaskSetDb(SStreamMeta* pMeta, void* arg, char* key) {
} }
STaskDbWrapper* pBackend = taskDbOpen(pMeta->path, key, chkpId); STaskDbWrapper* pBackend = taskDbOpen(pMeta->path, key, chkpId);
if (pBackend == NULL) { while (1) {
taosThreadMutexUnlock(&pMeta->backendMutex); if (pBackend == NULL) {
return -1; taosThreadMutexUnlock(&pMeta->backendMutex);
taosMsleep(1000);
stDebug("backed holded by other task, restart later, path: %s, key: %s", pMeta->path, key);
} else {
taosThreadMutexUnlock(&pMeta->backendMutex);
break;
}
taosThreadMutexLock(&pMeta->backendMutex);
pBackend = taskDbOpen(pMeta->path, key, chkpId);
} }
int64_t tref = taosAddRef(taskDbWrapperId, pBackend); int64_t tref = taosAddRef(taskDbWrapperId, pBackend);
@ -794,7 +802,7 @@ static void doClear(void* pKey, void* pVal, TBC* pCur, SArray* pRecycleList) {
} }
int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) { int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta) {
TBC* pCur = NULL; TBC* pCur = NULL;
void* pKey = NULL; void* pKey = NULL;
int32_t kLen = 0; int32_t kLen = 0;
void* pVal = NULL; void* pVal = NULL;