diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index ae48f2fe29848e6ee9ab3ddd19ff87707cd627c8..0c822a4007e1309e9ded0523157da8990d3a4537 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -223,27 +223,12 @@ static FORCE_INLINE void* streamQueueCurItem(SStreamQueue* queue) { return queue->qItem; } -static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) { - int8_t dequeueFlag = atomic_exchange_8(&queue->status, STREAM_QUEUE__PROCESSING); - if (dequeueFlag == STREAM_QUEUE__FAILED) { - ASSERT(queue->qItem != NULL); - return streamQueueCurItem(queue); - } else { - queue->qItem = NULL; - taosGetQitem(queue->qall, &queue->qItem); - if (queue->qItem == NULL) { - taosReadAllQitems(queue->queue, queue->qall); - taosGetQitem(queue->qall, &queue->qItem); - } - return streamQueueCurItem(queue); - } -} +void* streamQueueNextItem(SStreamQueue* queue); SStreamDataSubmit2* streamDataSubmitNew(SPackedData submit); +void streamDataSubmitDestroy(SStreamDataSubmit2* pDataSubmit); -void streamDataSubmitRefDec(SStreamDataSubmit2* pDataSubmit); - -SStreamDataSubmit2* streamSubmitRefClone(SStreamDataSubmit2* pSubmit); +SStreamDataSubmit2* streamSubmitBlockClone(SStreamDataSubmit2* pSubmit); typedef struct { char* qmsg; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index ae98589d6cbf328a3ad5759232436dd8de703a8a..5f171eb5eec58e6de05013903ecc41c69b426652 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1431,7 +1431,7 @@ int32_t tqProcessSubmitReq(STQ* pTq, SPackedData submit) { } if (pSubmit) { - streamDataSubmitRefDec(pSubmit); + streamDataSubmitDestroy(pSubmit); taosFreeQitem(pSubmit); } diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index c10469b13ed5c8468bce2a586a58f79f66507048..62c80c4ce4dbcc7d963e032b1bf635e519ca3956 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -30,7 +30,7 @@ static int32_t tqLoopExecFromQueue(STQ* pTq, STqHandle* pHandle, SStreamDataSubm // update processed atomic_store_64(&pHandle->pushHandle.processedVer, pSubmit->ver); streamQueueProcessSuccess(&pHandle->pushHandle.inputQ); - streamDataSubmitRefDec(pSubmit); + streamDataSubmitDestroy(pSubmit); if (pRsp->blockNum > 0) { *ppSubmit = pSubmit; return 0; @@ -58,7 +58,7 @@ int32_t tqExecFromInputQ(STQ* pTq, STqHandle* pHandle) { } while (pHandle->pushHandle.processedVer > pSubmit->ver + 1) { streamQueueProcessSuccess(&pHandle->pushHandle.inputQ); - streamDataSubmitRefDec(pSubmit); + streamDataSubmitDestroy(pSubmit); pSubmit = streamQueueNextItem(&pHandle->pushHandle.inputQ); if (pSubmit == NULL) break; } @@ -120,7 +120,7 @@ int32_t tqPreparePush(STQ* pTq, STqHandle* pHandle, int64_t reqId, const SRpcHan int32_t tqEnqueue(STqHandle* pHandle, SStreamDataSubmit* pSubmit) { int8_t inputStatus = atomic_load_8(&pHandle->pushHandle.inputStatus); if (inputStatus == TASK_INPUT_STATUS__NORMAL) { - SStreamDataSubmit* pSubmitClone = streamSubmitRefClone(pSubmit); + SStreamDataSubmit* pSubmitClone = streamSubmitBlockClone(pSubmit); if (pSubmitClone == NULL) { return -1; } diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index b4f2600f3080dd0fe977c4deebbc69431df9aaf0..d87d445ba570663015dc054dc0b76b87e7ce512c 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -279,18 +279,18 @@ int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem* pItem) { int8_t type = pItem->type; if (type == STREAM_INPUT__DATA_SUBMIT) { - SStreamDataSubmit2* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit2*)pItem); - if (pSubmitClone == NULL) { + SStreamDataSubmit2* pSubmitBlock = streamSubmitBlockClone((SStreamDataSubmit2*)pItem); + if (pSubmitBlock == NULL) { qDebug("task %d %p submit enqueue failed since out of memory", pTask->taskId, pTask); terrno = TSDB_CODE_OUT_OF_MEMORY; atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED); return -1; } - taosWriteQitem(pTask->inputQueue->queue, pSubmitClone); + taosWriteQitem(pTask->inputQueue->queue, pSubmitBlock); qDebug("stream task:%d %p submit enqueue %p %p %p msgLen:%d ver:%" PRId64 ", total in queue:%d", pTask->taskId, - pTask, pItem, pSubmitClone, pSubmitClone->submit.msgStr, pSubmitClone->submit.msgLen, - pSubmitClone->submit.ver, pTask->inputQueue->queue->numOfItems); + pTask, pItem, pSubmitBlock, pSubmitBlock->submit.msgStr, pSubmitBlock->submit.msgLen, + pSubmitBlock->submit.ver, pTask->inputQueue->queue->numOfItems); } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE || type == STREAM_INPUT__REF_DATA_BLOCK) { taosWriteQitem(pTask->inputQueue->queue, pItem); @@ -309,4 +309,20 @@ int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem* pItem) { atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__NORMAL); #endif return 0; +} + +void* streamQueueNextItem(SStreamQueue* queue) { + int8_t dequeueFlag = atomic_exchange_8(&queue->status, STREAM_QUEUE__PROCESSING); + if (dequeueFlag == STREAM_QUEUE__FAILED) { + ASSERT(queue->qItem != NULL); + return streamQueueCurItem(queue); + } else { + queue->qItem = NULL; + taosGetQitem(queue->qall, &queue->qItem); + if (queue->qItem == NULL) { + taosReadAllQitems(queue->queue, queue->qall); + taosGetQitem(queue->qall, &queue->qItem); + } + return streamQueueCurItem(queue); + } } \ No newline at end of file diff --git a/source/libs/stream/src/streamData.c b/source/libs/stream/src/streamData.c index ddddc1ceb3c2dbba99357b6b40381a3f8fce7c09..3fba1cb556a259b290dcca8764499d3d589e1da8 100644 --- a/source/libs/stream/src/streamData.c +++ b/source/libs/stream/src/streamData.c @@ -48,10 +48,12 @@ int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock if (pArray == NULL) { return -1; } + taosArrayPush(pArray, &(SSDataBlock){0}); SRetrieveTableRsp* pRetrieve = pReq->pRetrieve; SSDataBlock* pDataBlock = taosArrayGet(pArray, 0); blockDecode(pDataBlock, pRetrieve->data); + // TODO: refactor pDataBlock->info.window.skey = be64toh(pRetrieve->skey); pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey); @@ -79,25 +81,40 @@ SStreamDataSubmit2* streamDataSubmitNew(SPackedData submit) { } pDataSubmit->submit = submit; - *pDataSubmit->dataRef = 1; + *pDataSubmit->dataRef = 1; // initialize the reference count to be 1 pDataSubmit->type = STREAM_INPUT__DATA_SUBMIT; + return pDataSubmit; } +void streamDataSubmitDestroy(SStreamDataSubmit2* pDataSubmit) { + int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1); + ASSERT(ref >= 0 && pDataSubmit->type == STREAM_INPUT__DATA_SUBMIT); + + if (ref == 0) { + taosMemoryFree(pDataSubmit->submit.msgStr); + taosMemoryFree(pDataSubmit->dataRef); + } +} + SStreamMergedSubmit2* streamMergedSubmitNew() { SStreamMergedSubmit2* pMerged = (SStreamMergedSubmit2*)taosAllocateQitem(sizeof(SStreamMergedSubmit2), DEF_QITEM, 0); + if (pMerged == NULL) { + return NULL; + } - if (pMerged == NULL) return NULL; pMerged->submits = taosArrayInit(0, sizeof(SPackedData)); pMerged->dataRefs = taosArrayInit(0, sizeof(void*)); - if (pMerged->dataRefs == NULL || pMerged->submits == NULL) goto FAIL; + + if (pMerged->dataRefs == NULL || pMerged->submits == NULL) { + taosArrayDestroy(pMerged->submits); + taosArrayDestroy(pMerged->dataRefs); + taosFreeQitem(pMerged); + return NULL; + } + pMerged->type = STREAM_INPUT__MERGED_SUBMIT; return pMerged; -FAIL: - if (pMerged->submits) taosArrayDestroy(pMerged->submits); - if (pMerged->dataRefs) taosArrayDestroy(pMerged->dataRefs); - taosFreeQitem(pMerged); - return NULL; } int32_t streamMergeSubmit(SStreamMergedSubmit2* pMerged, SStreamDataSubmit2* pSubmit) { @@ -111,26 +128,17 @@ static FORCE_INLINE void streamDataSubmitRefInc(SStreamDataSubmit2* pDataSubmit) atomic_add_fetch_32(pDataSubmit->dataRef, 1); } -SStreamDataSubmit2* streamSubmitRefClone(SStreamDataSubmit2* pSubmit) { +SStreamDataSubmit2* streamSubmitBlockClone(SStreamDataSubmit2* pSubmit) { SStreamDataSubmit2* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit2), DEF_QITEM, 0); - if (pSubmitClone == NULL) { return NULL; } + streamDataSubmitRefInc(pSubmit); memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit2)); return pSubmitClone; } -void streamDataSubmitRefDec(SStreamDataSubmit2* pDataSubmit) { - int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1); - ASSERT(ref >= 0); - if (ref == 0) { - taosMemoryFree(pDataSubmit->submit.msgStr); - taosMemoryFree(pDataSubmit->dataRef); - } -} - SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* elem) { ASSERT(elem); if (dst->type == STREAM_INPUT__DATA_BLOCK && elem->type == STREAM_INPUT__DATA_BLOCK) { @@ -168,7 +176,7 @@ void streamFreeQitem(SStreamQueueItem* data) { taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)blockDataFreeRes); taosFreeQitem(data); } else if (type == STREAM_INPUT__DATA_SUBMIT) { - streamDataSubmitRefDec((SStreamDataSubmit2*)data); + streamDataSubmitDestroy((SStreamDataSubmit2*)data); taosFreeQitem(data); } else if (type == STREAM_INPUT__MERGED_SUBMIT) { SStreamMergedSubmit2* pMerge = (SStreamMergedSubmit2*)data;