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

enh: schedule vnodeCommit uniformly distributed

This commit is contained in:
Benguang Zhao 2022-12-27 16:28:44 +08:00
parent 691b75adf5
commit 92e05b8ece
5 changed files with 20 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/vnode/src/inc/vnd.h
View File

@ -86,6 +86,7 @@ int32_t vnodeGetBatchMeta(SVnode* pVnode, SRpcMsg* pMsg);
// vnodeCommit.c // vnodeCommit.c
int32_t vnodeBegin(SVnode* pVnode); int32_t vnodeBegin(SVnode* pVnode);
int32_t vnodeShouldCommit(SVnode* pVnode); int32_t vnodeShouldCommit(SVnode* pVnode);
void vnodeUpdCommitSched(SVnode* pVnode);
void vnodeRollback(SVnode* pVnode); void vnodeRollback(SVnode* pVnode);
int32_t vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg); int32_t vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg);
int32_t vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo); int32_t vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo);

7
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -332,6 +332,11 @@ struct STsdbKeepCfg {
int32_t keep2; int32_t keep2;
}; };
typedef struct SVCommitSched {
int64_t commitMs;
int64_t maxWaitMs;
} SVCommitSched;
struct SVnode { struct SVnode {
char* path; char* path;
SVnodeCfg config; SVnodeCfg config;
@ -350,7 +355,7 @@ struct SVnode {
STQ* pTq; STQ* pTq;
SSink* pSink; SSink* pSink;
tsem_t canCommit; tsem_t canCommit;
int64_t commitMs; SVCommitSched commitSched;
int64_t sync; int64_t sync;
TdThreadMutex lock; TdThreadMutex lock;
bool blocked; bool blocked;

13
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -58,15 +58,22 @@ int vnodeBegin(SVnode *pVnode) {
return 0; return 0;
} }
void vnodeUpdCommitSched(SVnode *pVnode) {
int64_t randNum = taosRand();
pVnode->commitSched.commitMs = taosGetMonoTimestampMs();
pVnode->commitSched.maxWaitMs = SYNC_VND_COMMIT_MAX_MS + (randNum % SYNC_VND_COMMIT_MAX_MS);
}
int vnodeShouldCommit(SVnode *pVnode) { int vnodeShouldCommit(SVnode *pVnode) {
if (!pVnode->inUse || !osDataSpaceAvailable()) { if (!pVnode->inUse || !osDataSpaceAvailable()) {
return false; return false;
} }
SVCommitSched *pSched = &pVnode->commitSched;
int64_t nowMs = taosGetMonoTimestampMs(); int64_t nowMs = taosGetMonoTimestampMs();
return (((pVnode->inUse->size > pVnode->inUse->node.size) && (pVnode->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) || return (((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) ||
(pVnode->inUse->size > 0 && pVnode->commitMs + SYNC_VND_COMMIT_MAX_MS < nowMs)); (pVnode->inUse->size > 0 && pSched->commitMs + pSched->maxWaitMs < nowMs));
} }
int vnodeShouldCommitOld(SVnode *pVnode) { int vnodeShouldCommitOld(SVnode *pVnode) {
@ -306,7 +313,7 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
vInfo("vgId:%d, start to commit, commitId:%" PRId64 " version:%" PRId64 " term: %" PRId64, TD_VID(pVnode), vInfo("vgId:%d, start to commit, commitId:%" PRId64 " version:%" PRId64 " term: %" PRId64, TD_VID(pVnode),
pInfo->info.state.commitID, pInfo->info.state.committed, pInfo->info.state.commitTerm); pInfo->info.state.commitID, pInfo->info.state.committed, pInfo->info.state.commitTerm);
pVnode->commitMs = taosGetMonoTimestampMs(); vnodeUpdCommitSched(pVnode);
// persist wal before starting // persist wal before starting
if (walPersist(pVnode->pWal) < 0) { if (walPersist(pVnode->pWal) < 0) {

3
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -142,7 +142,6 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
pVnode->path = (char *)&pVnode[1]; pVnode->path = (char *)&pVnode[1];
strcpy(pVnode->path, path); strcpy(pVnode->path, path);
pVnode->config = info.config; pVnode->config = info.config;
pVnode->commitMs = taosGetMonoTimestampMs();
pVnode->state.committed = info.state.committed; pVnode->state.committed = info.state.committed;
pVnode->state.commitTerm = info.state.commitTerm; pVnode->state.commitTerm = info.state.commitTerm;
pVnode->state.commitID = info.state.commitID; pVnode->state.commitID = info.state.commitID;
@ -158,6 +157,8 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
taosThreadMutexInit(&pVnode->mutex, NULL); taosThreadMutexInit(&pVnode->mutex, NULL);
taosThreadCondInit(&pVnode->poolNotEmpty, NULL); taosThreadCondInit(&pVnode->poolNotEmpty, NULL);
vnodeUpdCommitSched(pVnode);
int8_t rollback = vnodeShouldRollback(pVnode); int8_t rollback = vnodeShouldRollback(pVnode);
// open buffer pool // open buffer pool

2
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -154,7 +154,7 @@ void vnodeProposeCommitOnNeed(SVnode *pVnode) {
vInfo("vgId:%d, proposed vnode commit", pVnode->config.vgId); vInfo("vgId:%d, proposed vnode commit", pVnode->config.vgId);
_out: _out:
pVnode->commitMs = taosGetMonoTimestampMs(); vnodeUpdCommitSched(pVnode);
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
rpcMsg.pCont = NULL; rpcMsg.pCont = NULL;
} }