From 81f821a6fcb0f8a6167e0c12c07eaa993253d4d1 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 18 May 2023 10:51:43 +0800 Subject: [PATCH] refactor: do some internal refactor. --- include/common/tcommon.h | 5 - source/dnode/vnode/inc/vnode.h | 2 +- source/dnode/vnode/src/inc/vnodeInt.h | 2 - source/dnode/vnode/src/tq/tq.c | 153 -------------------------- source/dnode/vnode/src/tq/tqPush.c | 38 ++++++- source/dnode/vnode/src/tq/tqRead.c | 22 ++-- source/dnode/vnode/src/tq/tqRestore.c | 73 ++++++------ 7 files changed, 89 insertions(+), 206 deletions(-) diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 0544247da2..4faea6e5e3 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -208,11 +208,6 @@ typedef struct SSDataBlock { SDataBlockInfo info; } SSDataBlock; -enum { - FETCH_TYPE__DATA = 0, - FETCH_TYPE__NONE, -}; - typedef struct SVarColAttr { int32_t* offset; // start position for each entry in the list uint32_t length; // used buffer size that contain the valid data diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 549173b3f7..a2fe0bef21 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -267,7 +267,7 @@ int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList); int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList); int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id); -int32_t tqNextBlockInWal(STqReader* pReader); +bool tqNextBlockInWal(STqReader* pReader, const char* idstr); bool tqNextBlockImpl(STqReader *pReader, const char* idstr); int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, const char* id); diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 861235a5c3..230e8b8f88 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -219,8 +219,6 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t version, char* msg, int32_t msgL int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen); int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessStreamTaskCheckRsp(STQ* pTq, int64_t version, char* msg, int32_t msgLen); -int32_t tqProcessSubmitReqForSubscribe(STQ* pTq); -int32_t tqProcessDeleteDataReq(STQ* pTq, void* pReq, int32_t len, int64_t ver); int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec); int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 0179480011..75e25b26b1 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1195,159 +1195,6 @@ int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStream return TSDB_CODE_SUCCESS; } -int32_t tqProcessDeleteDataReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) { - bool failed = false; - SDecoder* pCoder = &(SDecoder){0}; - SDeleteRes* pRes = &(SDeleteRes){0}; - - pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t)); - if (pRes->uidList == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - failed = true; - } - - tDecoderInit(pCoder, pReq, len); - tDecodeDeleteRes(pCoder, pRes); - tDecoderClear(pCoder); - - int32_t sz = taosArrayGetSize(pRes->uidList); - if (sz == 0 || pRes->affectedRows == 0) { - taosArrayDestroy(pRes->uidList); - return 0; - } - - SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA); - blockDataEnsureCapacity(pDelBlock, sz); - pDelBlock->info.rows = sz; - pDelBlock->info.version = ver; - - for (int32_t i = 0; i < sz; i++) { - // start key column - SColumnInfoData* pStartCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX); - colDataSetVal(pStartCol, i, (const char*)&pRes->skey, false); // end key column - SColumnInfoData* pEndCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX); - colDataSetVal(pEndCol, i, (const char*)&pRes->ekey, false); - // uid column - SColumnInfoData* pUidCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX); - int64_t* pUid = taosArrayGet(pRes->uidList, i); - colDataSetVal(pUidCol, i, (const char*)pUid, false); - - colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, GROUPID_COLUMN_INDEX), i); - colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), i); - colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), i); - } - - taosArrayDestroy(pRes->uidList); - - int32_t* pRef = taosMemoryMalloc(sizeof(int32_t)); - *pRef = 1; - - taosWLockLatch(&pTq->pStreamMeta->lock); - - void* pIter = NULL; - while (1) { - pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter); - if (pIter == NULL) { - break; - } - - SStreamTask* pTask = *(SStreamTask**)pIter; - if (pTask->taskLevel != TASK_LEVEL__SOURCE) { - continue; - } - - qDebug("s-task:%s delete req enqueue, ver: %" PRId64, pTask->id.idStr, ver); - - if (!failed) { - SStreamRefDataBlock* pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0); - pRefBlock->type = STREAM_INPUT__REF_DATA_BLOCK; - pRefBlock->pBlock = pDelBlock; - - if (tAppendDataToInputQueue(pTask, (SStreamQueueItem*)pRefBlock) < 0) { - atomic_sub_fetch_32(pRef, 1); - taosFreeQitem(pRefBlock); - continue; - } - - if (streamSchedExec(pTask) < 0) { - qError("s-task:%s stream task launch failed", pTask->id.idStr); - continue; - } - - } else { - streamTaskInputFail(pTask); - } - } - - taosWUnLockLatch(&pTq->pStreamMeta->lock); - - int32_t ref = atomic_sub_fetch_32(pRef, 1); - if (ref == 0) { - blockDataDestroy(pDelBlock); - taosMemoryFree(pRef); - } - -#if 0 - SStreamDataBlock* pStreamBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0); - pStreamBlock->type = STREAM_INPUT__DATA_BLOCK; - pStreamBlock->blocks = taosArrayInit(0, sizeof(SSDataBlock)); - SSDataBlock block = {0}; - assignOneDataBlock(&block, pDelBlock); - block.info.type = STREAM_DELETE_DATA; - taosArrayPush(pStreamBlock->blocks, &block); - - if (!failed) { - if (tAppendDataToInputQueue(pTask, (SStreamQueueItem*)pStreamBlock) < 0) { - qError("stream task input del failed, task id %d", pTask->id.taskId); - continue; - } - - if (streamSchedExec(pTask) < 0) { - qError("stream task launch failed, task id %d", pTask->id.taskId); - continue; - } - } else { - streamTaskInputFail(pTask); - } - } - blockDataDestroy(pDelBlock); -#endif - return 0; -} - -int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) { - int32_t vgId = TD_VID(pTq->pVnode); - - taosWLockLatch(&pTq->lock); - - if (taosHashGetSize(pTq->pPushMgr) > 0) { - void* pIter = taosHashIterate(pTq->pPushMgr, NULL); - - while (pIter) { - STqHandle* pHandle = *(STqHandle**)pIter; - tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId); - - if (ASSERT(pHandle->msg != NULL)) { - tqError("pHandle->msg should not be null"); - break; - }else{ - SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info}; - tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg); - taosMemoryFree(pHandle->msg); - pHandle->msg = NULL; - } - - pIter = taosHashIterate(pTq->pPushMgr, pIter); - } - - taosHashClear(pTq->pPushMgr); - } - - // unlock - taosWUnLockLatch(&pTq->lock); - return 0; -} - int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) { SStreamTaskRunReq* pReq = pMsg->pCont; diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index 82a76657fd..7f4fe48b8e 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -16,8 +16,40 @@ #include "tq.h" #include "vnd.h" -int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) { +int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) { + int32_t vgId = TD_VID(pTq->pVnode); + + taosWLockLatch(&pTq->lock); + + if (taosHashGetSize(pTq->pPushMgr) > 0) { + void* pIter = taosHashIterate(pTq->pPushMgr, NULL); + + while (pIter) { + STqHandle* pHandle = *(STqHandle**)pIter; + tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId); + if (ASSERT(pHandle->msg != NULL)) { + tqError("pHandle->msg should not be null"); + break; + }else{ + SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info}; + tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg); + taosMemoryFree(pHandle->msg); + pHandle->msg = NULL; + } + + pIter = taosHashIterate(pTq->pPushMgr, pIter); + } + + taosHashClear(pTq->pPushMgr); + } + + // unlock + taosWUnLockLatch(&pTq->lock); + return 0; +} + +int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) { if (msgType == TDMT_VND_SUBMIT) { tqProcessSubmitReqForSubscribe(pTq); } @@ -37,10 +69,6 @@ int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t v if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_DELETE) { tqStartStreamTasks(pTq); } - - if (msgType == TDMT_VND_DELETE) { -// tqProcessDeleteDataReq(pTq, POINTER_SHIFT(msg, sizeof(SMsgHead)), msgLen - sizeof(SMsgHead), ver); - } } return 0; diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index 6bf09b703b..9e245fb45d 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -344,7 +344,7 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, const char* id) { } // todo ignore the error in wal? -int32_t tqNextBlockInWal(STqReader* pReader) { +bool tqNextBlockInWal(STqReader* pReader, const char* id) { SWalReader* pWalReader = pReader->pWalReader; while (1) { @@ -353,7 +353,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) { // try next message in wal file // todo always retry to avoid read failure caused by wal file deletion if (walNextValidMsg(pWalReader) < 0) { - return FETCH_TYPE__NONE; + return false; } void* pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg)); @@ -379,7 +379,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) { if (tDecodeSubmitReq(&decoder, &pReader->submit) < 0) { tDecoderClear(&decoder); tqError("decode wal file error, msgLen:%d, ver:%" PRId64, bodyLen, ver); - return FETCH_TYPE__NONE; + return false; } tDecoderClear(&decoder); @@ -388,15 +388,15 @@ int32_t tqNextBlockInWal(STqReader* pReader) { size_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData); while (pReader->nextBlk < numOfBlocks) { - tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg.msgStr, pReader->msg.msgLen, - pReader->msg.ver, pReader->nextBlk); + tqDebug("tq reader next data block %d/%d, len:%d %" PRId64 " %d", pReader->nextBlk, + numOfBlocks, pReader->msg.msgLen, pReader->msg.ver, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); if (pReader->tbIdHash == NULL) { int32_t code = tqRetrieveDataBlock(pReader, NULL); if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) { - return FETCH_TYPE__DATA; + return true; } } @@ -406,7 +406,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) { int32_t code = tqRetrieveDataBlock(pReader, NULL); if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) { - return FETCH_TYPE__DATA; + return true; } } else { pReader->nextBlk += 1; @@ -414,7 +414,9 @@ int32_t tqNextBlockInWal(STqReader* pReader) { } } + qDebug("stream scan return empty, all %d submit blocks consumed, %s", numOfBlocks, id); tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); + pReader->msg.msgStr = NULL; } } @@ -729,13 +731,17 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, const char* id) { return 0; } +// todo refactor: int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) { tqDebug("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); pReader->nextBlk++; - if (pSubmitTbDataRet) *pSubmitTbDataRet = pSubmitTbData; + if (pSubmitTbDataRet) { + *pSubmitTbDataRet = pSubmitTbData; + } + int32_t sversion = pSubmitTbData->sver; int64_t suid = pSubmitTbData->suid; int64_t uid = pSubmitTbData->uid; diff --git a/source/dnode/vnode/src/tq/tqRestore.c b/source/dnode/vnode/src/tq/tqRestore.c index 09e4ca4093..9acf2454af 100644 --- a/source/dnode/vnode/src/tq/tqRestore.c +++ b/source/dnode/vnode/src/tq/tqRestore.c @@ -15,7 +15,7 @@ #include "tq.h" -static int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle); +static int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle); // this function should be executed by stream threads. // extract submit block from WAL, and add them into the input queue for the sources tasks. @@ -30,7 +30,7 @@ int32_t tqStreamTasksScanWal(STQ* pTq) { // check all restore tasks bool shouldIdle = true; - createStreamRunReq(pTq->pStreamMeta, &shouldIdle); + createStreamTaskRunReq(pTq->pStreamMeta, &shouldIdle); int32_t times = 0; @@ -57,7 +57,39 @@ int32_t tqStreamTasksScanWal(STQ* pTq) { return 0; } -int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) { +static int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) { + // seek the stored version and extract data from WAL + int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader); + if (pTask->chkInfo.currentVer < firstVer) { + pTask->chkInfo.currentVer = firstVer; + tqWarn("vgId:%d s-task:%s ver earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64, vgId, + pTask->id.idStr, firstVer, pTask->chkInfo.currentVer); + + // todo need retry if failed + int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer); + if (code != TSDB_CODE_SUCCESS) { + return code; + } + + // append the data for the stream + tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer); + } else { + int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader); + if (currentVer == -1) { // we only seek the read for the first time + int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer); + if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit + return code; + } + + // append the data for the stream + tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer); + } + } + + return TSDB_CODE_SUCCESS; +} + +int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) { *pScanIdle = true; bool noNewDataInWal = true; int32_t vgId = pStreamMeta->vgId; @@ -67,6 +99,7 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) { return TSDB_CODE_SUCCESS; } + // clone the task list, to avoid the task update during scan wal files SArray* pTaskList = NULL; taosWLockLatch(&pStreamMeta->lock); pTaskList = taosArrayDup(pStreamMeta->pTaskList, NULL); @@ -107,38 +140,15 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) { *pScanIdle = false; // seek the stored version and extract data from WAL - int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader); - if (pTask->chkInfo.currentVer < firstVer) { - pTask->chkInfo.currentVer = firstVer; - tqWarn("vgId:%d s-task:%s ver earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64, vgId, - pTask->id.idStr, firstVer, pTask->chkInfo.currentVer); - - // todo need retry if failed - int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer); - if (code != TSDB_CODE_SUCCESS) { - streamMetaReleaseTask(pStreamMeta, pTask); - continue; - } - - // append the data for the stream - tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer); - } else { - int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader); - if (currentVer == -1) { - int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer); - if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit - streamMetaReleaseTask(pStreamMeta, pTask); - continue; - } - - // append the data for the stream - tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer); - } + int32_t code = doSetOffsetForWalReader(pTask, vgId); + if (code != TSDB_CODE_SUCCESS) { + streamMetaReleaseTask(pStreamMeta, pTask); + continue; } // append the data for the stream SStreamQueueItem* pItem = NULL; - int32_t code = extractMsgFromWal(pTask->exec.pWalReader, (void**) &pItem, pTask->id.idStr); + code = extractMsgFromWal(pTask->exec.pWalReader, (void**) &pItem, pTask->id.idStr); if (code != TSDB_CODE_SUCCESS) { // failed, continue streamMetaReleaseTask(pStreamMeta, pTask); continue; @@ -161,7 +171,6 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) { tqError("s-task:%s append input queue failed, ver:%" PRId64, pTask->id.idStr, pTask->chkInfo.currentVer); } - streamMetaReleaseTask(pStreamMeta, pTask); } -- GitLab