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

Merge pull request #27263 from taosdata/fix/3_liaohj 

fix(stream): delay to load the stream task.
This commit is contained in:
Haojun Liao 2024-08-16 09:11:30 +08:00 committed by GitHub
parent 111ec91e59 868898d11c
commit 7c44227048
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 12 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mnode/impl/inc/mndStream.h
View File

@ -154,8 +154,8 @@ bool streamTaskIterNextTask(SStreamTaskIter *pIter);
int32_t streamTaskIterGetCurrent(SStreamTaskIter *pIter, SStreamTask **pTask); int32_t streamTaskIterGetCurrent(SStreamTaskIter *pIter, SStreamTask **pTask);
int32_t mndInitExecInfo(); int32_t mndInitExecInfo();
void mndInitStreamExecInfo(SMnode *pMnode, SStreamExecInfo *pExecInfo); void mndInitStreamExecInfo(SMnode *pMnode, SStreamExecInfo *pExecInfo);
void mndInitStreamExecInfoForLeader(SMnode *pMnode); void mndStreamResetInitTaskListLoadFlag();
void mndInitStreamExecInfoUpdateRole(SMnode *pMnode, int32_t role); void mndUpdateStreamExecInfoRole(SMnode *pMnode, int32_t role);
int32_t removeExpiredNodeEntryAndTaskInBuf(SArray *pNodeSnapshot); int32_t removeExpiredNodeEntryAndTaskInBuf(SArray *pNodeSnapshot);
void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode); void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode);

15
source/dnode/mnode/impl/src/mndStream.c
View File

@ -2886,13 +2886,12 @@ void mndInitStreamExecInfo(SMnode *pMnode, SStreamExecInfo *pExecInfo) {
pExecInfo->initTaskList = true; pExecInfo->initTaskList = true;
} }
void mndInitStreamExecInfoForLeader(SMnode* pMnode) { void mndStreamResetInitTaskListLoadFlag() {
mInfo("reset task list buffer init flag for leader");
execInfo.initTaskList = false; execInfo.initTaskList = false;
mInfo("init stream execInfo for leader");
mndInitStreamExecInfo(pMnode, &execInfo);
} }
void mndInitStreamExecInfoUpdateRole(SMnode* pMnode, int32_t role) { void mndUpdateStreamExecInfoRole(SMnode* pMnode, int32_t role) {
execInfo.switchFromFollower = false; execInfo.switchFromFollower = false;
if (execInfo.role == NODE_ROLE_UNINIT) { if (execInfo.role == NODE_ROLE_UNINIT) {
@ -3017,7 +3016,8 @@ static int32_t mndProcessDropOrphanTaskReq(SRpcMsg *pReq) {
// check if it is conflict with other trans in both sourceDb and targetDb. // check if it is conflict with other trans in both sourceDb and targetDb.
bool conflict = mndStreamTransConflictCheck(pMnode, pTask->streamId, MND_STREAM_DROP_NAME, false); bool conflict = mndStreamTransConflictCheck(pMnode, pTask->streamId, MND_STREAM_DROP_NAME, false);
if (conflict) { if (conflict) {
TAOS_RETURN(TSDB_CODE_MND_TRANS_CONFLICT); code = TSDB_CODE_MND_TRANS_CONFLICT;
goto _err;
} }
SStreamObj dummyObj = {.uid = pTask->streamId, .sourceDb = "", .targetSTbName = ""}; SStreamObj dummyObj = {.uid = pTask->streamId, .sourceDb = "", .targetSTbName = ""};
@ -3030,8 +3030,7 @@ static int32_t mndProcessDropOrphanTaskReq(SRpcMsg *pReq) {
code = mndStreamRegisterTrans(pTrans, MND_STREAM_DROP_NAME, pTask->streamId); code = mndStreamRegisterTrans(pTrans, MND_STREAM_DROP_NAME, pTask->streamId);
if (code) { if (code) {
mndTransDrop(pTrans); goto _err;
return code;
} }
// drop all tasks // drop all tasks
@ -3055,7 +3054,7 @@ _err:
tDestroyDropOrphanTaskMsg(&msg); tDestroyDropOrphanTaskMsg(&msg);
mndTransDrop(pTrans); mndTransDrop(pTrans);
if (code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS || code == TSDB_CODE_ACTION_IN_PROGRESS) {
mDebug("create drop %d orphan tasks trans succ", numOfTasks); mDebug("create drop %d orphan tasks trans succ", numOfTasks);
} }
return code; return code;

6
source/dnode/mnode/impl/src/mndSync.c
View File

@ -364,7 +364,7 @@ static void mndBecomeFollower(const SSyncFSM *pFsm) {
} }
(void)taosThreadMutexUnlock(&pMgmt->lock); (void)taosThreadMutexUnlock(&pMgmt->lock);
mndInitStreamExecInfoUpdateRole(pMnode, NODE_ROLE_FOLLOWER); mndUpdateStreamExecInfoRole(pMnode, NODE_ROLE_FOLLOWER);
} }
static void mndBecomeLearner(const SSyncFSM *pFsm) { static void mndBecomeLearner(const SSyncFSM *pFsm) {
@ -388,8 +388,8 @@ static void mndBecomeLeader(const SSyncFSM *pFsm) {
mInfo("vgId:1, become leader"); mInfo("vgId:1, become leader");
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
mndInitStreamExecInfoUpdateRole(pMnode, NODE_ROLE_LEADER); mndUpdateStreamExecInfoRole(pMnode, NODE_ROLE_LEADER);
mndInitStreamExecInfoForLeader(pMnode); mndStreamResetInitTaskListLoadFlag();
} }
static bool mndApplyQueueEmpty(const SSyncFSM *pFsm) { static bool mndApplyQueueEmpty(const SSyncFSM *pFsm) {