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

refact task backend

This commit is contained in:
yihaoDeng 2023-10-07 17:03:12 +08:00
parent c146327da8
commit 3ebc7eef5f
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -715,9 +715,10 @@ int32_t delObsoleteCheckpoint(void* arg, const char* path) {
* chkpInUse is doing translation, cannot del until * chkpInUse is doing translation, cannot del until
* replication is finished * replication is finished
*/ */
int32_t chkpDelObsolete(void* arg, char* path) { int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
STaskBackendWrapper* pBackend = arg; STaskBackendWrapper* pBackend = arg;
taosThreadRwlockWrlock(&pBackend->chkpDirLock); taosThreadRwlockWrlock(&pBackend->chkpDirLock);
taosArrayPush(pBackend->chkpSaved, &chkpId);
SArray* chkpDel = taosArrayInit(8, sizeof(int64_t)); SArray* chkpDel = taosArrayInit(8, sizeof(int64_t));
SArray* chkpDup = taosArrayInit(8, sizeof(int64_t)); SArray* chkpDup = taosArrayInit(8, sizeof(int64_t));
@ -1041,9 +1042,9 @@ int32_t taskBackendDoCheckpoint(void* arg, uint64_t chkpId) {
qError("stream backend:%p failed to flush db at:%s", pBackend, pChkpIdDir); qError("stream backend:%p failed to flush db at:%s", pBackend, pChkpIdDir);
} }
code = chkpDelObsolete(pBackend, pChkpDir); code = chkpMayDelObsolete(pBackend, chkpId, pChkpDir);
taosReleaseRef(taskBackendWrapperId, refId);
taosReleaseRef(taskBackendWrapperId, refId);
return code; return code;
_EXIT: _EXIT: