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

enh: propose vnode commit synchronously

This commit is contained in:
Benguang Zhao 2023-04-20 10:21:50 +08:00
parent 55c6f115b0
commit a21e0fe75e
6 changed files with 2 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsg.h
View File

@ -77,7 +77,7 @@ static inline bool tmsgIsValid(tmsg_t type) {
} }
static inline bool vnodeIsMsgBlock(tmsg_t type) { static inline bool vnodeIsMsgBlock(tmsg_t type) {
return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) || return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) ||
(type == TDMT_VND_UPDATE_TAG_VAL) || (type == TDMT_VND_ALTER_CONFIRM); (type == TDMT_VND_UPDATE_TAG_VAL) || (type == TDMT_VND_ALTER_CONFIRM) || (type == TDMT_VND_COMMIT);
} }
static inline bool syncUtilUserCommit(tmsg_t msgType) { static inline bool syncUtilUserCommit(tmsg_t msgType) {

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

@ -97,7 +97,6 @@ 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, bool atExit); int32_t vnodeShouldCommit(SVnode* pVnode, bool atExit);
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); int32_t vnodeCommitInfo(const char* dir);

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

@ -378,7 +378,6 @@ struct SVnode {
STQ* pTq; STQ* pTq;
SSink* pSink; SSink* pSink;
tsem_t canCommit; tsem_t canCommit;
SVCommitSched commitSched;
int64_t sync; int64_t sync;
TdThreadMutex lock; TdThreadMutex lock;
bool blocked; bool blocked;
@ -387,9 +386,6 @@ struct SVnode {
int32_t blockSec; int32_t blockSec;
int64_t blockSeq; int64_t blockSeq;
SQHandle* pQuery; SQHandle* pQuery;
#if 0
SRpcHandleInfo blockInfo;
#endif
}; };
#define TD_VID(PVNODE) ((PVNODE)->config.vgId) #define TD_VID(PVNODE) ((PVNODE)->config.vgId)

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

@ -143,23 +143,13 @@ _exit:
return code; return code;
} }
void vnodeUpdCommitSched(SVnode *pVnode) {
int64_t randNum = taosRand();
pVnode->commitSched.commitMs = taosGetMonoTimestampMs();
pVnode->commitSched.maxWaitMs = tsVndCommitMaxIntervalMs + (randNum % tsVndCommitMaxIntervalMs);
}
int vnodeShouldCommit(SVnode *pVnode, bool atExit) { int vnodeShouldCommit(SVnode *pVnode, bool atExit) {
SVCommitSched *pSched = &pVnode->commitSched;
int64_t nowMs = taosGetMonoTimestampMs();
bool diskAvail = osDataSpaceAvailable(); bool diskAvail = osDataSpaceAvailable();
bool needCommit = false; bool needCommit = false;
taosThreadMutexLock(&pVnode->mutex); taosThreadMutexLock(&pVnode->mutex);
if (pVnode->inUse && diskAvail) { if (pVnode->inUse && diskAvail) {
needCommit = needCommit = (pVnode->inUse->size > pVnode->inUse->node.size) || (pVnode->inUse->size > 0 && atExit);
((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) ||
((pVnode->inUse->size > 0) && atExit);
} }
taosThreadMutexUnlock(&pVnode->mutex); taosThreadMutexUnlock(&pVnode->mutex);
return needCommit; return needCommit;
@ -431,8 +421,6 @@ 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);
vnodeUpdCommitSched(pVnode);
// persist wal before starting // persist wal before starting
if (walPersist(pVnode->pWal) < 0) { if (walPersist(pVnode->pWal) < 0) {
vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), terrstr()); vError("vgId:%d, failed to persist wal since %s", TD_VID(pVnode), terrstr());

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

@ -286,8 +286,6 @@ 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

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

@ -112,9 +112,6 @@ static int32_t inline vnodeProposeMsg(SVnode *pVnode, SRpcMsg *pMsg, bool isWeak
pVnode->blocked = true; pVnode->blocked = true;
pVnode->blockSec = taosGetTimestampSec(); pVnode->blockSec = taosGetTimestampSec();
pVnode->blockSeq = seq; pVnode->blockSeq = seq;
#if 0
pVnode->blockInfo = pMsg->info;
#endif
} }
taosThreadMutexUnlock(&pVnode->lock); taosThreadMutexUnlock(&pVnode->lock);
@ -157,8 +154,6 @@ void vnodeProposeCommitOnNeed(SVnode *pVnode, bool atExit) {
} else { } else {
tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &rpcMsg); tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &rpcMsg);
} }
vnodeUpdCommitSched(pVnode);
} }
#if BATCH_ENABLE #if BATCH_ENABLE