提交 a6e3da80 编写于 作者: C Cary Xu

chore: code optimization for sma commit

上级 4c6623d3
...@@ -206,9 +206,9 @@ int32_t smaBegin(SSma* pSma); ...@@ -206,9 +206,9 @@ int32_t smaBegin(SSma* pSma);
int32_t smaSyncPreCommit(SSma* pSma); int32_t smaSyncPreCommit(SSma* pSma);
int32_t smaSyncCommit(SSma* pSma); int32_t smaSyncCommit(SSma* pSma);
int32_t smaSyncPostCommit(SSma* pSma); int32_t smaSyncPostCommit(SSma* pSma);
int32_t smaAsyncPreCommit(SSma* pSma); int32_t smaPreCommit(SSma* pSma);
int32_t smaAsyncCommit(SSma* pSma); int32_t smaCommit(SSma* pSma);
int32_t smaAsyncPostCommit(SSma* pSma); int32_t smaPostCommit(SSma* pSma);
int32_t smaDoRetention(SSma* pSma, int64_t now); int32_t smaDoRetention(SSma* pSma, int64_t now);
int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg); int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
......
...@@ -54,28 +54,28 @@ int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(p ...@@ -54,28 +54,28 @@ int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(p
#endif #endif
/** /**
* @brief Only applicable to Rollup SMA * @brief async commit, only applicable to Rollup SMA
* *
* @param pSma * @param pSma
* @return int32_t * @return int32_t
*/ */
int32_t smaAsyncPreCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma); } int32_t smaPreCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma); }
/** /**
* @brief Only applicable to Rollup SMA * @brief async commit, only applicable to Rollup SMA
* *
* @param pSma * @param pSma
* @return int32_t * @return int32_t
*/ */
int32_t smaAsyncCommit(SSma *pSma) { return tdProcessRSmaAsyncCommitImpl(pSma); } int32_t smaCommit(SSma *pSma) { return tdProcessRSmaAsyncCommitImpl(pSma); }
/** /**
* @brief Only applicable to Rollup SMA * @brief async commit, only applicable to Rollup SMA
* *
* @param pSma * @param pSma
* @return int32_t * @return int32_t
*/ */
int32_t smaAsyncPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); } int32_t smaPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }
/** /**
* @brief set rsma trigger stat active * @brief set rsma trigger stat active
...@@ -366,9 +366,11 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) { ...@@ -366,9 +366,11 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
* @return int32_t * @return int32_t
*/ */
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) { static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
int32_t code = 0;
SVnode *pVnode = pSma->pVnode;
SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma); SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
if (!pSmaEnv) { if (!pSmaEnv) {
return TSDB_CODE_SUCCESS; goto _exit;
} }
#if 0 #if 0
SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv); SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
...@@ -378,8 +380,21 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) { ...@@ -378,8 +380,21 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
#endif #endif
if ((code = tsdbCommit(VND_RSMA0(pVnode))) < 0) {
return TSDB_CODE_SUCCESS; smaError("vgId:%d, failed to commit tsdb rsma0 since %s", TD_VID(pVnode), tstrerror(code));
goto _exit;
}
if ((code = tsdbCommit(VND_RSMA1(pVnode))) < 0) {
smaError("vgId:%d, failed to commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(code));
goto _exit;
}
if ((code = tsdbCommit(VND_RSMA2(pVnode))) < 0) {
smaError("vgId:%d, failed to commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(code));
goto _exit;
}
_exit:
terrno = code;
return code;
} }
/** /**
......
...@@ -239,10 +239,8 @@ int vnodeCommit(SVnode *pVnode) { ...@@ -239,10 +239,8 @@ int vnodeCommit(SVnode *pVnode) {
} }
walBeginSnapshot(pVnode->pWal, pVnode->state.applied); walBeginSnapshot(pVnode->pWal, pVnode->state.applied);
// preCommit if (smaPreCommit(pVnode->pSma) < 0) {
// smaSyncPreCommit(pVnode->pSma); vError("vgId:%d, failed to pre-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
if(smaAsyncPreCommit(pVnode->pSma) < 0){
vError("vgId:%d, failed to async pre-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
return -1; return -1;
} }
...@@ -256,21 +254,8 @@ int vnodeCommit(SVnode *pVnode) { ...@@ -256,21 +254,8 @@ int vnodeCommit(SVnode *pVnode) {
} }
if (VND_IS_RSMA(pVnode)) { if (VND_IS_RSMA(pVnode)) {
if (smaAsyncCommit(pVnode->pSma) < 0) { if (smaCommit(pVnode->pSma) < 0) {
vError("vgId:%d, failed to async commit sma since %s", TD_VID(pVnode), tstrerror(terrno)); vError("vgId:%d, failed to commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
}
if (tsdbCommit(VND_RSMA0(pVnode)) < 0) {
vError("vgId:%d, failed to commit tsdb rsma0 since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
}
if (tsdbCommit(VND_RSMA1(pVnode)) < 0) {
vError("vgId:%d, failed to commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
}
if (tsdbCommit(VND_RSMA2(pVnode)) < 0) {
vError("vgId:%d, failed to commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(terrno));
return -1; return -1;
} }
} else { } else {
...@@ -294,10 +279,8 @@ int vnodeCommit(SVnode *pVnode) { ...@@ -294,10 +279,8 @@ int vnodeCommit(SVnode *pVnode) {
pVnode->state.committed = info.state.committed; pVnode->state.committed = info.state.committed;
// postCommit if (smaPostCommit(pVnode->pSma) < 0) {
// smaSyncPostCommit(pVnode->pSma); vError("vgId:%d, failed to post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
if (smaAsyncPostCommit(pVnode->pSma) < 0) {
vError("vgId:%d, failed to async post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
return -1; return -1;
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册