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

Merge pull request #20967 from taosdata/fix/liaohj_main 

fix(stream): update the reference count value to be int32
This commit is contained in:
Haojun Liao 2023-04-18 19:29:29 +08:00 committed by GitHub
parent 2af63992ea fb32ffd960
commit 915e5acb87
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/stream/tstream.h
View File

@ -345,7 +345,7 @@ typedef struct SStreamMeta {
FTaskExpand* expandFunc;
int32_t vgId;
SRWLatch lock;
int8_t walScan;
int32_t walScan;
} SStreamMeta;
int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);

24
source/dnode/vnode/src/tq/tqRestore.c
View File

@ -27,31 +27,37 @@ int tqStreamTasksScanWal(STQ* pTq) {
int64_t st = taosGetTimestampMs();
while (1) {
tqInfo("vgId:%d continue check if data in wal are available", vgId);
int32_t scan = pMeta->walScan;
tqDebug("vgId:%d continue check if data in wal are available, scan:%d", vgId, scan);
// check all restore tasks
bool allFull = true;
streamTaskReplayWal(pTq->pStreamMeta, pTq->pOffsetStore, &allFull);
bool shouldIdle = true;
streamTaskReplayWal(pTq->pStreamMeta, pTq->pOffsetStore, &shouldIdle);
int32_t times = 0;
if (allFull) {
if (shouldIdle) {
taosWLockLatch(&pMeta->lock);
pMeta->walScan -= 1;
times = pMeta->walScan;
ASSERT(pMeta->walScan >= 0);
if (pMeta->walScan <= 0) {
taosWUnLockLatch(&pMeta->lock);
break;
}
taosWUnLockLatch(&pMeta->lock);
tqInfo("vgId:%d scan wal for stream tasks for %d times", vgId, times);
tqDebug("vgId:%d scan wal for stream tasks for %d times", vgId, times);
}
}
double el = (taosGetTimestampMs() - st) / 1000.0;
tqInfo("vgId:%d scan wal for stream tasks completed, elapsed time:%.2f sec", vgId, el);
int64_t el = (taosGetTimestampMs() - st);
tqDebug("vgId:%d scan wal for stream tasks completed, elapsed time:%"PRId64" ms", vgId, el);
// restore wal scan flag
// atomic_store_8(&pTq->pStreamMeta->walScan, 0);
return 0;
}
@ -96,8 +102,8 @@ int32_t streamTaskReplayWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetSto
continue;
}
int8_t status = pTask->status.taskStatus;
if (status == TASK_STATUS__RECOVER_PREPARE || status == TASK_STATUS__WAIT_DOWNSTREAM) {
if (pTask->status.taskStatus == TASK_STATUS__RECOVER_PREPARE ||
pTask->status.taskStatus == TASK_STATUS__WAIT_DOWNSTREAM) {
tqDebug("s-task:%s skip push data, not ready for processing, status %d", pTask->id.idStr,
pTask->status.taskStatus);
continue;