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

enh: rsma exec logic optimization

This commit is contained in:
Cary Xu 2022-09-19 11:16:20 +08:00
parent eda8968252
commit ab49ce6eaf
2 changed files with 10 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -295,12 +295,17 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
}
/**
* @brief step 3: consume the SubmitReq in buffer
* @brief step 3: commit should wait for all SubmitReq in buffer be consumed
* 1) This is high cost task and should not put in asyncPreCommit originally.
* 2) But, if put in asyncCommit, would trigger taskInfo cloning frequently.
*/
if (tdRSmaProcessExecImpl(pSma, RSMA_EXEC_COMMIT) < 0) {
return TSDB_CODE_FAILED;
nLoops = 0;
while (atomic_load_64(&pRSmaStat->nBufItems) > 0) {
++nLoops;
if (nLoops > 1000) {
sched_yield();
nLoops = 0;
}
}
smaInfo("vgId:%d, rsma commit, wait for all items to be consumed, TID:%p", SMA_VID(pSma),

39
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -1822,7 +1822,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
while (true) {
// step 1: rsma exec - consume data in buffer queue for all suids
if (type == RSMA_EXEC_OVERFLOW || type == RSMA_EXEC_COMMIT) {
if (type == RSMA_EXEC_OVERFLOW) {
void *pIter = NULL;
while ((pIter = taosHashIterate(infoHash, pIter))) {
SRSmaInfo *pInfo = *(SRSmaInfo **)pIter;
@ -1878,42 +1878,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
atomic_val_compare_exchange_8(&pInfo->assigned, 1, 0);
}
}
if (type == RSMA_EXEC_COMMIT) {
if (atomic_load_64(&pRSmaStat->nBufItems) <= 0) {
break;
} else {
// commit should wait for all items be consumed
continue;
}
}
}
#if 0
else if (type == RSMA_EXEC_COMMIT) {
while (pIter) {
SRSmaInfo *pInfo = *(SRSmaInfo **)pIter;
if (taosQueueItemSize(pInfo->iQueue)) {
if (atomic_val_compare_exchange_8(&pInfo->assigned, 0, 1) == 0) {
taosReadAllQitems(pInfo->iQueue, pInfo->iQall); // queue has mutex lock
int32_t qallItemSize = taosQallItemSize(pInfo->iQall);
if (qallItemSize > 0) {
atomic_fetch_sub_64(&pRSmaStat->nBufItems, qallItemSize);
nIdle = 0;
// batch exec
tdRSmaBatchExec(pSma, pInfo, pInfo->qall, pSubmitArr, type);
}
// tdRSmaFetchAllResult(pSma, pInfo, pSubmitArr);
atomic_val_compare_exchange_8(&pInfo->assigned, 1, 0);
}
}
ASSERT(taosQueueItemSize(pInfo->iQueue) == 0);
pIter = taosHashIterate(infoHash, pIter);
}
break;
}
#endif
else {
} else {
ASSERT(0);
}