From 5531a5a121155cf2b28bbadda404a7e0e5d62e31 Mon Sep 17 00:00:00 2001 From: Cary Xu Date: Tue, 30 Aug 2022 11:33:34 +0800 Subject: [PATCH] enh: rsam fetch all logic optimization --- source/dnode/vnode/src/inc/sma.h | 1 + source/dnode/vnode/src/sma/smaRollup.c | 9 +++++++-- source/dnode/vnode/src/sma/smaSnapshot.c | 2 +- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/source/dnode/vnode/src/inc/sma.h b/source/dnode/vnode/src/inc/sma.h index abfffc045f..c29c4cb6c4 100644 --- a/source/dnode/vnode/src/inc/sma.h +++ b/source/dnode/vnode/src/inc/sma.h @@ -95,6 +95,7 @@ struct SRSmaStat { int64_t refId; // shared by fetch tasks volatile int64_t nBufItems; // number of items in queue buffer SRWLatch lock; // r/w lock for rsma fs(e.g. qtaskinfo) + volatile int32_t nFetchAll; // active number of fetch all int8_t triggerStat; // shared by fetch tasks int8_t commitStat; // 0 not in committing, 1 in committing SArray *aTaskFile; // qTaskFiles committed recently(for recovery/snapshot r/w) diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index 52b08d131c..89cdd58c4e 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -1714,9 +1714,14 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) { smaDebug("vgId:%d, batchSize:%d, execType:%" PRIi8, SMA_VID(pSma), qallItemSize, type); } - if (atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 2) == 0) { + int8_t oldStat = atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 2); + if (oldStat == 0 || + ((oldStat == 2) && atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)) < TASK_TRIGGER_STAT_PAUSED)) { + atomic_fetch_add_32(&pRSmaStat->nFetchAll, 1); tdRSmaFetchAllResult(pSma, pInfo, pSubmitArr); - atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0); + if (0 == atomic_sub_fetch_32(&pRSmaStat->nFetchAll, 1)) { + atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0); + } } if (qallItemSize > 0) { diff --git a/source/dnode/vnode/src/sma/smaSnapshot.c b/source/dnode/vnode/src/sma/smaSnapshot.c index 335c15a539..28d7218bc8 100644 --- a/source/dnode/vnode/src/sma/smaSnapshot.c +++ b/source/dnode/vnode/src/sma/smaSnapshot.c @@ -80,7 +80,7 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapRead goto _err; } pReader->pQTaskFReader->pReadH = qTaskF; -#if 0 +#if 1 SQTaskFile* pQTaskF = &pReader->pQTaskFReader->fTask; pQTaskF->nRef = 1; #endif -- GitLab