diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 842e656b9ded148534e1cc488dff53071eb95ae1..0dd137d5327b5992d740a3e18e01e0320b599596 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -226,11 +226,36 @@ typedef struct { int32_t nodeId; int32_t childId; int32_t taskId; - int64_t checkpointVer; - int64_t processedVer; - SEpSet epSet; + // int64_t checkpointVer; + // int64_t processedVer; + SEpSet epSet; } SStreamChildEpInfo; +typedef struct { + int32_t nodeId; + int32_t childId; + int64_t stateSaveVer; + int64_t stateProcessedVer; +} SStreamCheckpointInfo; + +typedef struct { + int64_t streamId; + int64_t checkTs; + int32_t checkpointId; // incremental + int32_t taskId; + SArray* checkpointVer; // SArray +} SStreamMultiVgCheckpointInfo; + +typedef struct { + int32_t taskId; + int32_t checkpointId; // incremental +} SStreamCheckpointKey; + +typedef struct { + int32_t taskId; + SArray* checkpointVer; +} SStreamRecoveringState; + typedef struct SStreamTask { int64_t streamId; int32_t taskId; @@ -256,6 +281,8 @@ typedef struct SStreamTask { // children info SArray* childEpInfo; // SArray + int32_t nextCheckId; + SArray* checkpointInfo; // SArray // exec STaskExec exec; @@ -445,6 +472,7 @@ typedef struct { int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq); int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq); +void tFreeStreamDispatchReq(SStreamDispatchReq* pReq); int32_t streamSetupTrigger(SStreamTask* pTask); @@ -468,6 +496,7 @@ typedef struct SStreamMeta { TTB* pTaskDb; TTB* pStateDb; SHashObj* pTasks; + SHashObj* pRecoveringState; void* ahandle; TXN txn; FTaskExpand* expandFunc; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index c1c680fc56df0ff76f5c1c9c13897bfd233cf34c..f7da287ca12e11a34c6a64077c30fe3235d99ae2 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -859,8 +859,10 @@ void vnodeEnqueueStreamMsg(SVnode* pVnode, SRpcMsg* pMsg) { tDecoderInit(&decoder, msgBody, msgLen); if (tDecodeStreamDispatchReq(&decoder, &req) < 0) { code = TSDB_CODE_MSG_DECODE_ERROR; + tDecoderClear(&decoder); goto FAIL; } + tDecoderClear(&decoder); int32_t taskId = req.taskId; diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index 30f0919ceeaedc52912189ec7150cb9c48d0000e..e4bf90f3088b33e2e6188e153f527a3b7cb845ec 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -136,6 +136,7 @@ int32_t streamTaskEnqueue(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pRsp->pCont = buf; pRsp->contLen = sizeof(SMsgHead) + sizeof(SStreamDispatchRsp); tmsgSendRsp(pRsp); + tFreeStreamDispatchReq(pReq); return status == TASK_INPUT_STATUS__NORMAL ? 0 : -1; } diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 8d6d31e37f1f13f41c5f7d969f6040a68fb02a15..8b5e6355a22179bc3548c8231a7978c73e26f089 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -62,6 +62,12 @@ int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) { return 0; } +void tFreeStreamDispatchReq(SStreamDispatchReq* pReq) { + taosArrayDestroyP(pReq->data, taosMemoryFree); + taosArrayDestroy(pReq->dataLen); + taosMemoryFree(pReq); +} + int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq) { if (tStartEncode(pEncoder) < 0) return -1; if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1; diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index e662c18a15e269aa31131841c10b1328e55a2e98..4a4d67b89aa03ae86017cf6dea94a5a274b5aae6 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -15,7 +15,7 @@ #include "streamInc.h" -static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) { +static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* pRes) { void* exec = pTask->exec.executor; // set input @@ -82,14 +82,16 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) return 0; } +#if 0 static FORCE_INLINE int32_t streamUpdateVer(SStreamTask* pTask, SStreamDataBlock* pBlock) { ASSERT(pBlock->type == STREAM_INPUT__DATA_BLOCK); int32_t childId = pBlock->childId; int64_t ver = pBlock->sourceVer; SStreamChildEpInfo* pChildInfo = taosArrayGetP(pTask->childEpInfo, childId); - pChildInfo->processedVer = ver; + /*pChildInfo-> = ver;*/ return 0; } +#endif int32_t streamPipelineExec(SStreamTask* pTask, int32_t batchNum) { ASSERT(pTask->taskLevel != TASK_LEVEL__SINK); @@ -198,6 +200,8 @@ int32_t streamExecForAll(SStreamTask* pTask) { streamTaskExecImpl(pTask, data, pRes); qDebug("stream task %d exec end", pTask->taskId); + streamFreeQitem(data); + if (taosArrayGetSize(pRes) != 0) { SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); if (qRes == NULL) { diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c index 2a77ce8a91a0f6e416b32b1362f013238f99c82c..28693d15a6179636a81baa891b78338a7c4bd839 100644 --- a/source/libs/stream/src/streamRecover.c +++ b/source/libs/stream/src/streamRecover.c @@ -87,63 +87,95 @@ int32_t tDecodeSMStreamTaskRecoverRsp(SDecoder* pDecoder, SMStreamTaskRecoverRsp return 0; } -typedef struct { - int32_t vgId; - int32_t childId; - int64_t ver; -} SStreamVgVerCheckpoint; - -int32_t tEncodeSStreamVgVerCheckpoint(SEncoder* pEncoder, const SStreamVgVerCheckpoint* pCheckpoint) { - if (tEncodeI32(pEncoder, pCheckpoint->vgId) < 0) return -1; +int32_t tEncodeSStreamCheckpointInfo(SEncoder* pEncoder, const SStreamCheckpointInfo* pCheckpoint) { + if (tEncodeI32(pEncoder, pCheckpoint->nodeId) < 0) return -1; if (tEncodeI32(pEncoder, pCheckpoint->childId) < 0) return -1; - if (tEncodeI64(pEncoder, pCheckpoint->ver) < 0) return -1; + if (tEncodeI64(pEncoder, pCheckpoint->stateProcessedVer) < 0) return -1; return 0; } -int32_t tDecodeSStreamVgVerCheckpoint(SDecoder* pDecoder, SStreamVgVerCheckpoint* pCheckpoint) { - if (tDecodeI32(pDecoder, &pCheckpoint->vgId) < 0) return -1; +int32_t tDecodeSStreamCheckpointInfo(SDecoder* pDecoder, SStreamCheckpointInfo* pCheckpoint) { + if (tDecodeI32(pDecoder, &pCheckpoint->nodeId) < 0) return -1; if (tDecodeI32(pDecoder, &pCheckpoint->childId) < 0) return -1; - if (tDecodeI64(pDecoder, &pCheckpoint->ver) < 0) return -1; + if (tDecodeI64(pDecoder, &pCheckpoint->stateProcessedVer) < 0) return -1; return 0; } -typedef struct { - int64_t streamId; - int64_t checkTs; - int64_t checkpointId; - int32_t taskId; - SArray* checkpointVer; // SArray -} SStreamAggVerCheckpoint; - -int32_t tEncodeSStreamAggVerCheckpoint(SEncoder* pEncoder, const SStreamAggVerCheckpoint* pCheckpoint) { +int32_t tEncodeSStreamMultiVgCheckpointInfo(SEncoder* pEncoder, const SStreamMultiVgCheckpointInfo* pCheckpoint) { if (tEncodeI64(pEncoder, pCheckpoint->streamId) < 0) return -1; if (tEncodeI64(pEncoder, pCheckpoint->checkTs) < 0) return -1; - if (tEncodeI64(pEncoder, pCheckpoint->checkpointId) < 0) return -1; + if (tEncodeI32(pEncoder, pCheckpoint->checkpointId) < 0) return -1; if (tEncodeI32(pEncoder, pCheckpoint->taskId) < 0) return -1; int32_t sz = taosArrayGetSize(pCheckpoint->checkpointVer); if (tEncodeI32(pEncoder, sz) < 0) return -1; for (int32_t i = 0; i < sz; i++) { - SStreamVgVerCheckpoint* pOneVgCkpoint = taosArrayGet(pCheckpoint->checkpointVer, i); - if (tEncodeSStreamVgVerCheckpoint(pEncoder, pOneVgCkpoint) < 0) return -1; + SStreamCheckpointInfo* pOneVgCkpoint = taosArrayGet(pCheckpoint->checkpointVer, i); + if (tEncodeSStreamCheckpointInfo(pEncoder, pOneVgCkpoint) < 0) return -1; } return 0; } -int32_t tDecodeSStreamAggVerCheckpoint(SDecoder* pDecoder, SStreamAggVerCheckpoint* pCheckpoint) { +int32_t tDecodeSStreamMultiVgCheckpointInfo(SDecoder* pDecoder, SStreamMultiVgCheckpointInfo* pCheckpoint) { if (tDecodeI64(pDecoder, &pCheckpoint->streamId) < 0) return -1; if (tDecodeI64(pDecoder, &pCheckpoint->checkTs) < 0) return -1; - if (tDecodeI64(pDecoder, &pCheckpoint->checkpointId) < 0) return -1; + if (tDecodeI32(pDecoder, &pCheckpoint->checkpointId) < 0) return -1; if (tDecodeI32(pDecoder, &pCheckpoint->taskId) < 0) return -1; int32_t sz; if (tDecodeI32(pDecoder, &sz) < 0) return -1; for (int32_t i = 0; i < sz; i++) { - SStreamVgVerCheckpoint oneVgCheckpoint; - if (tDecodeSStreamVgVerCheckpoint(pDecoder, &oneVgCheckpoint) < 0) return -1; + SStreamCheckpointInfo oneVgCheckpoint; + if (tDecodeSStreamCheckpointInfo(pDecoder, &oneVgCheckpoint) < 0) return -1; taosArrayPush(pCheckpoint->checkpointVer, &oneVgCheckpoint); } return 0; } +int32_t streamCheckSinkLevel(SStreamMeta* pMeta, SStreamTask* pTask) { + void* buf = NULL; + + ASSERT(pTask->taskLevel == TASK_LEVEL__SINK); + int32_t sz = taosArrayGetSize(pTask->checkpointInfo); + + SStreamMultiVgCheckpointInfo checkpoint; + checkpoint.checkpointId = 0; + checkpoint.checkTs = taosGetTimestampMs(); + checkpoint.streamId = pTask->streamId; + checkpoint.taskId = pTask->taskId; + checkpoint.checkpointVer = pTask->checkpointInfo; + + int32_t len; + int32_t code; + tEncodeSize(tEncodeSStreamMultiVgCheckpointInfo, &checkpoint, len, code); + if (code < 0) { + return -1; + } + + buf = taosMemoryCalloc(1, len); + if (buf == NULL) { + return -1; + } + SEncoder encoder; + tEncoderInit(&encoder, buf, len); + tEncodeSStreamMultiVgCheckpointInfo(&encoder, &checkpoint); + tEncoderClear(&encoder); + + SStreamCheckpointKey key = { + .taskId = pTask->taskId, + .checkpointId = checkpoint.checkpointId, + }; + + if (tdbTbUpsert(pMeta->pStateDb, &key, sizeof(SStreamCheckpointKey), buf, len, &pMeta->txn) < 0) { + ASSERT(0); + goto FAIL; + } + + taosMemoryFree(buf); + return 0; +FAIL: + if (buf) taosMemoryFree(buf); + return -1; +} + int32_t streamRecoverSinkLevel(SStreamMeta* pMeta, SStreamTask* pTask) { ASSERT(pTask->taskLevel == TASK_LEVEL__SINK); // load status @@ -154,9 +186,39 @@ int32_t streamRecoverSinkLevel(SStreamMeta* pMeta, SStreamTask* pTask) { } SDecoder decoder; tDecoderInit(&decoder, pVal, vLen); - SStreamAggVerCheckpoint aggCheckpoint; - tDecodeSStreamAggVerCheckpoint(&decoder, &aggCheckpoint); - /*pTask->*/ + SStreamMultiVgCheckpointInfo aggCheckpoint; + tDecodeSStreamMultiVgCheckpointInfo(&decoder, &aggCheckpoint); + tDecoderClear(&decoder); + + pTask->nextCheckId = aggCheckpoint.checkpointId + 1; + pTask->checkpointInfo = aggCheckpoint.checkpointVer; + + return 0; +} + +int32_t streamCheckAggLevel(SStreamMeta* pMeta, SStreamTask* pTask) { + ASSERT(pTask->taskLevel == TASK_LEVEL__AGG); + // save and copy state + // save state info + return 0; +} + +int32_t streamRecoverAggLevel(SStreamMeta* pMeta, SStreamTask* pTask) { + ASSERT(pTask->taskLevel == TASK_LEVEL__AGG); + // try recover sink level + // after all sink level recovered, choose current state backend to recover + return 0; +} + +int32_t streamCheckSourceLevel(SStreamMeta* pMeta, SStreamTask* pTask) { + ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE); + // try recover agg level + // + return 0; +} + +int32_t streamRecoverSourceLevel(SStreamMeta* pMeta, SStreamTask* pTask) { + ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE); return 0; } diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 8b5bd849f6627802d8b3098efe0c4288f9ecf323..638d39e5cc386ccd5922936dc9606f9b3aa2d8cd 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -34,7 +34,7 @@ int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo) if (tEncodeI32(pEncoder, pInfo->taskId) < 0) return -1; if (tEncodeI32(pEncoder, pInfo->nodeId) < 0) return -1; if (tEncodeI32(pEncoder, pInfo->childId) < 0) return -1; - if (tEncodeI64(pEncoder, pInfo->processedVer) < 0) return -1; + /*if (tEncodeI64(pEncoder, pInfo->processedVer) < 0) return -1;*/ if (tEncodeSEpSet(pEncoder, &pInfo->epSet) < 0) return -1; return 0; } @@ -43,7 +43,7 @@ int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo) { if (tDecodeI32(pDecoder, &pInfo->taskId) < 0) return -1; if (tDecodeI32(pDecoder, &pInfo->nodeId) < 0) return -1; if (tDecodeI32(pDecoder, &pInfo->childId) < 0) return -1; - if (tDecodeI64(pDecoder, &pInfo->processedVer) < 0) return -1; + /*if (tDecodeI64(pDecoder, &pInfo->processedVer) < 0) return -1;*/ if (tDecodeSEpSet(pDecoder, &pInfo->epSet) < 0) return -1; return 0; }