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

feat: dispatch auto compact req

This commit is contained in:
kailixu 2024-12-04 17:50:17 +08:00
parent fa3b212980
commit 4707149369
1 changed files with 12 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
source/dnode/mnode/impl/src/mndCompact.c
View File

@ -891,58 +891,33 @@ static void mndCompactPullup(SMnode *pMnode) {
taosArrayDestroy(pArray); taosArrayDestroy(pArray);
} }
static int32_t mndProcessAutoCompact(SRpcMsg *pReq) { static int32_t mndCompactDispatch(SRpcMsg *pReq) {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
void *pDbIter = NULL; void *pDbIter = NULL;
while (1) { SDbObj *pDb = NULL;
SDbObj *pDb = NULL; while ((pDbIter = sdbFetch(pSdb, SDB_DB, pDbIter, (void **)&pDb))) {
pDbIter = sdbFetch(pSdb, SDB_DB, pDbIter, (void **)&pDb); if ((pDb->cfg.compactInterval == 0) || (pDb->cfg.compactStartTime >= pDb->cfg.compactEndTime)) {
if (pDbIter == NULL) break;
if ((pDb->cfg.compactInterval == 0) || (pDb->cfg.compactStartTime == pDb->cfg.compactEndTime)) {
sdbRelease(pSdb, pDb);
continue;
}
if (pDb->cfg.compactStartTime > pDb->cfg.compactEndTime) {
mWarn("db:%s compact start time is greater than end time", pDb->name);
sdbRelease(pSdb, pDb);
continue;
}
// check if there is an unfinished compact task for this db
bool hasCompact = false;
void *pCompactIter = NULL;
while (1) {
SCompactObj *pCompact = NULL;
pCompactIter = sdbFetch(pSdb, SDB_COMPACT, pCompactIter, (void **)&pCompact);
if (pCompactIter == NULL) break;
if (0 == strncmp(pCompact->dbname, pDb->name, TSDB_DB_FNAME_LEN) == 0) {
hasCompact = true;
sdbRelease(pSdb, pCompact);
break;
}
sdbRelease(pSdb, pCompact);
}
if (hasCompact) {
mInfo("db:%s skip auto compact since unfinished compact task", pDb->name);
sdbRelease(pSdb, pDb); sdbRelease(pSdb, pDb);
continue; continue;
} }
// mndCompactSendProgressReq(pMnode, pDb);
sdbRelease(pSdb, pDb); sdbRelease(pSdb, pDb);
} }
_exit:
return 0; if (code) {
mWarn("failed to dispatch auto compact at line %d since %s", lino, tstrerror(code));
}
return code;
} }
static int32_t mndProcessCompactTimer(SRpcMsg *pReq) { static int32_t mndProcessCompactTimer(SRpcMsg *pReq) {
mTrace("start to process compact timer"); mTrace("start to process compact timer");
mndCompactPullup(pReq->info.node); mndCompactPullup(pReq->info.node);
TAOS_UNUSED(mndProcessAutoCompact(pReq)); TAOS_UNUSED(mndCompactDispatch(pReq));
return 0; return 0;
} }