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

fix(stream): reset task add the failed chkptId info.

This commit is contained in:
Haojun Liao 2024-11-01 18:58:22 +08:00
parent 526904fdd1
commit e7b3a3aec8
9 changed files with 51 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/common/tmsg.h
View File

@ -3769,7 +3769,14 @@ typedef struct {
SMsgHead head; SMsgHead head;
int64_t streamId; int64_t streamId;
int32_t taskId; int32_t taskId;
} SVPauseStreamTaskReq, SVResetStreamTaskReq; } SVPauseStreamTaskReq;
typedef struct {
SMsgHead head;
int64_t streamId;
int32_t taskId;
int64_t chkptId;
} SVResetStreamTaskReq;
typedef struct { typedef struct {
char name[TSDB_STREAM_FNAME_LEN]; char name[TSDB_STREAM_FNAME_LEN];

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

@ -712,6 +712,7 @@ int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask);
void streamTaskSetTriggerDispatchConfirmed(SStreamTask* pTask, int32_t vgId); void streamTaskSetTriggerDispatchConfirmed(SStreamTask* pTask, int32_t vgId);
int32_t streamTaskSendCheckpointTriggerMsg(SStreamTask* pTask, int32_t dstTaskId, int32_t downstreamNodeId, int32_t streamTaskSendCheckpointTriggerMsg(SStreamTask* pTask, int32_t dstTaskId, int32_t downstreamNodeId,
SRpcHandleInfo* pInfo, int32_t code); SRpcHandleInfo* pInfo, int32_t code);
void streamTaskSetFailedCheckpointId(SStreamTask* pTask, int64_t failedId);
int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue); int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue);
int32_t streamQueueGetNumOfUnAccessedItems(const SStreamQueue* pQueue); int32_t streamQueueGetNumOfUnAccessedItems(const SStreamQueue* pQueue);

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

@ -56,6 +56,7 @@ typedef struct SStreamTransMgmt {
typedef struct SStreamTaskResetMsg { typedef struct SStreamTaskResetMsg {
int64_t streamId; int64_t streamId;
int32_t transId; int32_t transId;
int64_t checkpointId;
} SStreamTaskResetMsg; } SStreamTaskResetMsg;
typedef struct SChkptReportInfo { typedef struct SChkptReportInfo {
@ -141,9 +142,9 @@ int32_t mndStreamSetResumeAction(STrans *pTrans, SMnode *pMnode, SStreamObj *pSt
int32_t mndStreamSetPauseAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream); int32_t mndStreamSetPauseAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
int32_t mndStreamSetDropAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream); int32_t mndStreamSetDropAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
int32_t mndStreamSetDropActionFromList(SMnode *pMnode, STrans *pTrans, SArray *pList); int32_t mndStreamSetDropActionFromList(SMnode *pMnode, STrans *pTrans, SArray *pList);
int32_t mndStreamSetResetTaskAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream); int32_t mndStreamSetResetTaskAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream, int64_t chkptId);
int32_t mndStreamSetUpdateChkptAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream); int32_t mndStreamSetUpdateChkptAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream); int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream, int64_t chkptId);
int32_t mndStreamSetChkptIdAction(SMnode *pMnode, STrans *pTrans, SStreamTask* pTask, int64_t checkpointId, int64_t ts); int32_t mndStreamSetChkptIdAction(SMnode *pMnode, STrans *pTrans, SStreamTask* pTask, int64_t checkpointId, int64_t ts);
int32_t mndStreamSetRestartAction(SMnode* pMnode, STrans *pTrans, SStreamObj* pStream); int32_t mndStreamSetRestartAction(SMnode* pMnode, STrans *pTrans, SStreamObj* pStream);
int32_t mndStreamSetCheckpointAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTask, int64_t checkpointId, int32_t mndStreamSetCheckpointAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTask, int64_t checkpointId,

21
source/dnode/mnode/impl/src/mndStreamHb.c
View File

@ -24,7 +24,7 @@ typedef struct SFailedCheckpointInfo {
static int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode); static int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode);
static int32_t mndSendDropOrphanTasksMsg(SMnode *pMnode, SArray *pList); static int32_t mndSendDropOrphanTasksMsg(SMnode *pMnode, SArray *pList);
static int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId); static int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId, int64_t checkpointId);
static void updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage); static void updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage);
static void addIntoFailedChkptList(SArray *pList, const SFailedCheckpointInfo *pInfo); static void addIntoFailedChkptList(SArray *pList, const SFailedCheckpointInfo *pInfo);
static int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList); static int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList);
@ -68,7 +68,7 @@ void addIntoFailedChkptList(SArray *pList, const SFailedCheckpointInfo *pInfo) {
} }
} }
int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) { int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream, int64_t chkptId) {
STrans *pTrans = NULL; STrans *pTrans = NULL;
int32_t code = doCreateTrans(pMnode, pStream, NULL, TRN_CONFLICT_NOTHING, MND_STREAM_TASK_RESET_NAME, int32_t code = doCreateTrans(pMnode, pStream, NULL, TRN_CONFLICT_NOTHING, MND_STREAM_TASK_RESET_NAME,
" reset from failed checkpoint", &pTrans); " reset from failed checkpoint", &pTrans);
@ -84,7 +84,7 @@ int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
return code; return code;
} }
code = mndStreamSetResetTaskAction(pMnode, pTrans, pStream); code = mndStreamSetResetTaskAction(pMnode, pTrans, pStream, chkptId);
if (code) { if (code) {
sdbRelease(pMnode->pSdb, pStream); sdbRelease(pMnode->pSdb, pStream);
mndTransDrop(pTrans); mndTransDrop(pTrans);
@ -115,7 +115,7 @@ int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
return code; return code;
} }
int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId) { int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId, int64_t checkpointId) {
int32_t size = sizeof(SStreamTaskResetMsg); int32_t size = sizeof(SStreamTaskResetMsg);
int32_t num = taosArrayGetSize(execInfo.pKilledChkptTrans); int32_t num = taosArrayGetSize(execInfo.pKilledChkptTrans);
@ -135,8 +135,9 @@ int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t
taosArrayRemove(execInfo.pKilledChkptTrans, 0); // remove this first, append new reset trans in the tail taosArrayRemove(execInfo.pKilledChkptTrans, 0); // remove this first, append new reset trans in the tail
} }
SStreamTaskResetMsg p = {.streamId = streamId, .transId = transId}; SStreamTaskResetMsg p = {.streamId = streamId, .transId = transId, .checkpointId = checkpointId};
// let's remember that this trans had been killed already
void *px = taosArrayPush(execInfo.pKilledChkptTrans, &p); void *px = taosArrayPush(execInfo.pKilledChkptTrans, &p);
if (px == NULL) { if (px == NULL) {
mError("failed to push reset-msg trans:%d into the killed chkpt trans list, size:%d", transId, num - 1); mError("failed to push reset-msg trans:%d into the killed chkpt trans list, size:%d", transId, num - 1);
@ -150,6 +151,7 @@ int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t
pReq->streamId = streamId; pReq->streamId = streamId;
pReq->transId = transId; pReq->transId = transId;
pReq->checkpointId = checkpointId;
SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_TASK_RESET, .pCont = pReq, .contLen = size}; SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_TASK_RESET, .pCont = pReq, .contLen = size};
int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg); int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
@ -234,7 +236,7 @@ int32_t mndProcessResetStatusReq(SRpcMsg *pReq) {
} else { } else {
mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, transId:%d, create reset trans", pStream->name, mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, transId:%d, create reset trans", pStream->name,
pStream->uid, pMsg->transId); pStream->uid, pMsg->transId);
code = mndCreateStreamResetStatusTrans(pMnode, pStream); code = mndCreateStreamResetStatusTrans(pMnode, pStream, pMsg->checkpointId);
} }
} }
@ -495,10 +497,11 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) {
continue; continue;
} }
mInfo("checkpointId:%" PRId64 " transId:%d failed, issue task-reset trans to reset all tasks status", mInfo("stream:0x%" PRIx64 "checkpointId:%" PRId64
pInfo->checkpointId, pInfo->transId); " transId:%d failed issue task-reset trans to reset all tasks status",
pInfo->streamUid, pInfo->checkpointId, pInfo->transId);
code = mndSendResetFromCheckpointMsg(pMnode, pInfo->streamUid, pInfo->transId); code = mndSendResetFromCheckpointMsg(pMnode, pInfo->streamUid, pInfo->transId, pInfo->checkpointId);
if (code) { if (code) {
mError("failed to create reset task trans, code:%s", tstrerror(code)); mError("failed to create reset task trans, code:%s", tstrerror(code));
} }

7
source/dnode/mnode/impl/src/mndStreamTransAct.c
View File

@ -295,7 +295,7 @@ static int32_t doSetUpdateChkptAction(SMnode *pMnode, STrans *pTrans, SStreamTas
return code; return code;
} }
static int32_t doSetResetAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTask) { static int32_t doSetResetAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTask, int64_t chkptId) {
SVResetStreamTaskReq *pReq = taosMemoryCalloc(1, sizeof(SVResetStreamTaskReq)); SVResetStreamTaskReq *pReq = taosMemoryCalloc(1, sizeof(SVResetStreamTaskReq));
if (pReq == NULL) { if (pReq == NULL) {
mError("failed to malloc in reset stream, size:%" PRIzu ", code:%s", sizeof(SVResetStreamTaskReq), mError("failed to malloc in reset stream, size:%" PRIzu ", code:%s", sizeof(SVResetStreamTaskReq),
@ -306,6 +306,7 @@ static int32_t doSetResetAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTa
pReq->head.vgId = htonl(pTask->info.nodeId); pReq->head.vgId = htonl(pTask->info.nodeId);
pReq->taskId = pTask->id.taskId; pReq->taskId = pTask->id.taskId;
pReq->streamId = pTask->id.streamId; pReq->streamId = pTask->id.streamId;
pReq->chkptId = chkptId;
SEpSet epset = {0}; SEpSet epset = {0};
bool hasEpset = false; bool hasEpset = false;
@ -544,7 +545,7 @@ int32_t mndStreamSetDropActionFromList(SMnode *pMnode, STrans *pTrans, SArray* p
return 0; return 0;
} }
int32_t mndStreamSetResetTaskAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) { int32_t mndStreamSetResetTaskAction(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream, int64_t chkptId) {
SStreamTaskIter *pIter = NULL; SStreamTaskIter *pIter = NULL;
taosWLockLatch(&pStream->lock); taosWLockLatch(&pStream->lock);
@ -564,7 +565,7 @@ int32_t mndStreamSetResetTaskAction(SMnode *pMnode, STrans *pTrans, SStreamObj *
return code; return code;
} }
code = doSetResetAction(pMnode, pTrans, pTask); code = doSetResetAction(pMnode, pTrans, pTask, chkptId);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
destroyStreamTaskIter(pIter); destroyStreamTaskIter(pIter);
taosWUnLockLatch(&pStream->lock); taosWUnLockLatch(&pStream->lock);

12
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -935,7 +935,7 @@ int32_t tqStartTaskCompleteCallback(SStreamMeta* pMeta) {
} }
int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, char* pMsg) { int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, char* pMsg) {
SVPauseStreamTaskReq* pReq = (SVPauseStreamTaskReq*)pMsg; SVResetStreamTaskReq* pReq = (SVResetStreamTaskReq*)pMsg;
SStreamTask* pTask = NULL; SStreamTask* pTask = NULL;
int32_t code = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId, &pTask); int32_t code = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId, &pTask);
@ -950,17 +950,13 @@ int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, char* pMsg) {
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);
streamTaskClearCheckInfo(pTask, true); streamTaskClearCheckInfo(pTask, true);
streamTaskSetFailedCheckpointId(pTask, pReq->chkptId);
// clear flag set during do checkpoint, and open inputQ for all upstream tasks // clear flag set during do checkpoint, and open inputQ for all upstream tasks
SStreamTaskState pState = streamTaskGetStatus(pTask); SStreamTaskState pState = streamTaskGetStatus(pTask);
if (pState.state == TASK_STATUS__CK) { if (pState.state == TASK_STATUS__CK) {
int32_t tranId = 0;
int64_t activeChkId = 0;
streamTaskGetActiveCheckpointInfo(pTask, &tranId, &activeChkId);
tqDebug("s-task:%s reset task status from checkpoint, current checkpointingId:%" PRId64 ", transId:%d",
pTask->id.idStr, activeChkId, tranId);
streamTaskSetStatusReady(pTask); streamTaskSetStatusReady(pTask);
tqDebug("s-task:%s reset checkpoint status to ready", pTask->id.idStr);
} else if (pState.state == TASK_STATUS__UNINIT) { } else if (pState.state == TASK_STATUS__UNINIT) {
// tqDebug("s-task:%s start task by checking downstream tasks", pTask->id.idStr); // tqDebug("s-task:%s start task by checking downstream tasks", pTask->id.idStr);
// tqStreamTaskRestoreCheckpoint(pMeta, pTask->id.streamId, pTask->id.taskId); // tqStreamTaskRestoreCheckpoint(pMeta, pTask->id.streamId, pTask->id.taskId);

1
source/libs/stream/inc/streamInt.h
View File

@ -191,7 +191,6 @@ int32_t streamTaskSendCheckpointReadyMsg(SStreamTask* pTask);
int32_t streamTaskSendCheckpointSourceRsp(SStreamTask* pTask); int32_t streamTaskSendCheckpointSourceRsp(SStreamTask* pTask);
int32_t streamTaskSendCheckpointReq(SStreamTask* pTask); int32_t streamTaskSendCheckpointReq(SStreamTask* pTask);
void streamTaskSetFailedCheckpointId(SStreamTask* pTask);
int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask); int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask);
int32_t streamTaskGetNumOfUpstream(const SStreamTask* pTask); int32_t streamTaskGetNumOfUpstream(const SStreamTask* pTask);
int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*); int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*);

26
source/libs/stream/src/streamCheckpoint.c
View File

@ -560,7 +560,7 @@ void streamTaskClearCheckInfo(SStreamTask* pTask, bool clearChkpReadyMsg) {
} }
streamMutexUnlock(&pInfo->lock); streamMutexUnlock(&pInfo->lock);
stDebug("s-task:%s clear active checkpointInfo, failed checkpointId:%" PRId64 ", current checkpointId:%" PRId64, stDebug("s-task:%s clear active checkpointInfo, failed checkpointId:%" PRId64 ", latest checkpointId:%" PRId64,
pTask->id.idStr, pInfo->failedId, pTask->chkInfo.checkpointId); pTask->id.idStr, pInfo->failedId, pTask->chkInfo.checkpointId);
} }
@ -680,15 +680,22 @@ int32_t streamTaskUpdateTaskCheckpointInfo(SStreamTask* pTask, bool restored, SV
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void streamTaskSetFailedCheckpointId(SStreamTask* pTask) { void streamTaskSetFailedCheckpointId(SStreamTask* pTask, int64_t failedId) {
struct SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo; struct SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
if (pInfo->activeId <= 0) { if (failedId <= 0) {
stWarn("s-task:%s checkpoint-info is cleared now, not set the failed checkpoint info", pTask->id.idStr); stWarn("s-task:%s failedId is 0, not update the failed checkpoint info, current failedId:%" PRId64
" activeId:%" PRId64,
pTask->id.idStr, pInfo->failedId, pInfo->activeId);
} else { } else {
pInfo->failedId = pInfo->activeId; if (failedId <= pInfo->failedId) {
stDebug("s-task:%s mark and set the failed checkpointId:%" PRId64 " (transId:%d)", pTask->id.idStr, pInfo->activeId, stDebug("s-task:%s failedId:%" PRId64 " not update to:%" PRId64, pTask->id.idStr, pInfo->failedId, failedId);
pInfo->transId); } else {
stDebug("s-task:%s mark and set the failed checkpointId:%" PRId64 " (transId:%d) activeId:%" PRId64
" prev failedId:%" PRId64,
pTask->id.idStr, failedId, pInfo->transId, pInfo->activeId, pInfo->failedId);
pInfo->failedId = failedId;
}
} }
} }
@ -696,7 +703,7 @@ void streamTaskSetCheckpointFailed(SStreamTask* pTask) {
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);
ETaskStatus status = streamTaskGetStatus(pTask).state; ETaskStatus status = streamTaskGetStatus(pTask).state;
if (status == TASK_STATUS__CK) { if (status == TASK_STATUS__CK) {
streamTaskSetFailedCheckpointId(pTask); streamTaskSetFailedCheckpointId(pTask, pTask->chkInfo.pActiveInfo->activeId);
} }
streamMutexUnlock(&pTask->lock); streamMutexUnlock(&pTask->lock);
} }
@ -874,8 +881,9 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
code = streamSendChkptReportMsg(pTask, &pTask->chkInfo, dropRelHTask); code = streamSendChkptReportMsg(pTask, &pTask->chkInfo, dropRelHTask);
} }
} else { // clear the checkpoint info if failed } else { // clear the checkpoint info if failed
// set failed checkpoint id before clear the checkpoint info
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);
streamTaskSetFailedCheckpointId(pTask); // set failed checkpoint id before clear the checkpoint info streamTaskSetFailedCheckpointId(pTask, ckId);
streamMutexUnlock(&pTask->lock); streamMutexUnlock(&pTask->lock);
code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE); code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE);

4
source/libs/stream/src/streamTask.c
View File

@ -1248,13 +1248,13 @@ void streamTaskDestroyActiveChkptInfo(SActiveCheckpointInfo* pInfo) {
taosMemoryFree(pInfo); taosMemoryFree(pInfo);
} }
//NOTE: clear the checkpoint id, and keep the failed id // NOTE: clear the checkpoint id, and keep the failed id
// failedId for a task will increase as the checkpoint I.D. increases.
void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) { void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) {
pInfo->activeId = 0; pInfo->activeId = 0;
pInfo->transId = 0; pInfo->transId = 0;
pInfo->allUpstreamTriggerRecv = 0; pInfo->allUpstreamTriggerRecv = 0;
pInfo->dispatchTrigger = false; pInfo->dispatchTrigger = false;
// pInfo->failedId = 0;
taosArrayClear(pInfo->pDispatchTriggerList); taosArrayClear(pInfo->pDispatchTriggerList);
taosArrayClear(pInfo->pCheckpointReadyRecvList); taosArrayClear(pInfo->pCheckpointReadyRecvList);