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

refactor: do some internal refactor.

This commit is contained in:
Haojun Liao 2024-05-06 17:08:11 +08:00
parent 7ad7ef7cfc
commit c982f6ac80
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1441,7 +1441,7 @@ void streamMetaUpdateStageRole(SStreamMeta* pMeta, int64_t stage, bool isLeader)
} }
} }
static int32_t prepareBeforeStartTasks(SStreamMeta* pMeta, SArray** pList) { static int32_t prepareBeforeStartTasks(SStreamMeta* pMeta, SArray** pList, int64_t now) {
streamMetaWLock(pMeta); streamMetaWLock(pMeta);
if (pMeta->closeFlag) { if (pMeta->closeFlag) {
@ -1454,7 +1454,7 @@ static int32_t prepareBeforeStartTasks(SStreamMeta* pMeta, SArray** pList) {
taosHashClear(pMeta->startInfo.pReadyTaskSet); taosHashClear(pMeta->startInfo.pReadyTaskSet);
taosHashClear(pMeta->startInfo.pFailedTaskSet); taosHashClear(pMeta->startInfo.pFailedTaskSet);
pMeta->startInfo.startTs = taosGetTimestampMs(); pMeta->startInfo.startTs = now;
streamMetaResetTaskStatus(pMeta); streamMetaResetTaskStatus(pMeta);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
@ -1468,7 +1468,7 @@ int32_t streamMetaStartAllTasks(SStreamMeta* pMeta) {
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList); int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
stInfo("vgId:%d start to check all %d stream task(s) downstream status", vgId, numOfTasks); stInfo("vgId:%d start to check all %d stream task(s) downstream status, start ts:%"PRId64, vgId, numOfTasks, now);
if (numOfTasks == 0) { if (numOfTasks == 0) {
stInfo("vgId:%d no tasks to be started", pMeta->vgId); stInfo("vgId:%d no tasks to be started", pMeta->vgId);
@ -1476,7 +1476,7 @@ int32_t streamMetaStartAllTasks(SStreamMeta* pMeta) {
} }
SArray* pTaskList = NULL; SArray* pTaskList = NULL;
code = prepareBeforeStartTasks(pMeta, &pTaskList); code = prepareBeforeStartTasks(pMeta, &pTaskList, now);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
ASSERT(pTaskList == NULL); ASSERT(pTaskList == NULL);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;