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

enh: code optimization for insert_req statistics

This commit is contained in:
Cary Xu 2022-10-12 10:40:50 +08:00
parent c79fd325bd
commit 0691d7ff9a
1 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -819,11 +819,12 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
int32_t tsize, ret; int32_t tsize, ret;
SEncoder encoder = {0}; SEncoder encoder = {0};
SArray *newTbUids = NULL; SArray *newTbUids = NULL;
SVStatis statis = {0};
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
pRsp->code = 0; pRsp->code = 0;
pSubmitReq->version = version; pSubmitReq->version = version;
atomic_fetch_add_64(&pVnode->statis.nBatchInsert, 1); statis.nBatchInsert = 1;
#ifdef TD_DEBUG_PRINT_ROW #ifdef TD_DEBUG_PRINT_ROW
vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__); vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
@ -943,18 +944,21 @@ _exit:
taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp); taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp);
atomic_fetch_add_64(&pVnode->statis.nInsert, submitRsp.numOfRows);
atomic_fetch_add_64(&pVnode->statis.nInsertSuccess, submitRsp.affectedRows);
// TODO: the partial success scenario and the error case // TODO: the partial success scenario and the error case
// => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
// 1/level 2. // 1/level 2.
// TODO: refactor // TODO: refactor
if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) { if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
atomic_fetch_add_64(&pVnode->statis.nBatchInsertSuccess, 1); statis.nBatchInsertSuccess = 1;
tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT); tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);
} }
// N.B not strict as the following update steps is not atomic
atomic_add_fetch_64(&pVnode->statis.nInsert, submitRsp.numOfRows);
atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, submitRsp.affectedRows);
atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, statis.nBatchInsertSuccess);
atomic_add_fetch_64(&pVnode->statis.nBatchInsert, statis.nBatchInsert);
vDebug("vgId:%d, submit success, index:%" PRId64, pVnode->config.vgId, version); vDebug("vgId:%d, submit success, index:%" PRId64, pVnode->config.vgId, version);
return 0; return 0;
} }