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

[TD-5664]<feature> add more state of compact

This commit is contained in:
yihaoDeng 2021-07-29 03:27:58 +08:00
parent 3aa6acfc93
commit b022190279
5 changed files with 11 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/inc/tsdb.h
View File

@ -94,7 +94,7 @@ STsdbRepo *tsdbOpenRepo(STsdbCfg *pCfg, STsdbAppH *pAppH);
int tsdbCloseRepo(STsdbRepo *repo, int toCommit);
int32_t tsdbConfigRepo(STsdbRepo *repo, STsdbCfg *pCfg);
int tsdbGetState(STsdbRepo *repo);
bool tsdbInCompact(STsdbRepo *repo);
int8_t tsdbGetCompactState(STsdbRepo *repo);
// --------- TSDB TABLE DEFINITION
typedef struct {
uint64_t uid; // the unique table ID

4
src/tsdb/inc/tsdbint.h
View File

@ -92,7 +92,7 @@ struct STsdbRepo {
pthread_mutex_t mutex;
bool repoLocked;
int32_t code; // Commit code
bool inCompact; // is in compact process?
int8_t compactState; // compact state: inCompact/noCompact/waitingCompact?
};
#define REPO_ID(r) (r)->config.tsdbId
@ -139,4 +139,4 @@ static FORCE_INLINE int tsdbGetNextMaxTables(int tid) {
}
#endif
#endif /* _TD_TSDB_INT_H_ */
#endif /* _TD_TSDB_INT_H_ */

8
src/tsdb/src/tsdbCompact.c
View File

@ -58,6 +58,7 @@ static int tsdbCompactFSetImpl(SCompactH *pComph);
static int tsdbWriteBlockToRightFile(SCompactH *pComph, STable *pTable, SDataCols *pDataCols, void **ppBuf,
void **ppCBuf);
enum {TSDB_NO_COMPACT, TSDB_IN_COMPACT, TSDB_WAITING_COMPACT};
int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); }
void *tsdbCompactImpl(STsdbRepo *pRepo) {
@ -89,16 +90,17 @@ _err:
}
static int tsdbAsyncCompact(STsdbRepo *pRepo) {
pRepo->compactState = TSDB_WAITING_COMPACT;
tsem_wait(&(pRepo->readyToCommit));
return tsdbScheduleCommit(pRepo, COMPACT_REQ);
}
static void tsdbStartCompact(STsdbRepo *pRepo) {
ASSERT(!pRepo->inCompact);
assert(pRepo->compactState != TSDB_IN_COMPACT);
tsdbInfo("vgId:%d start to compact!", REPO_ID(pRepo));
tsdbStartFSTxn(pRepo, 0, 0);
pRepo->code = TSDB_CODE_SUCCESS;
pRepo->inCompact = true;
pRepo->compactState = TSDB_IN_COMPACT;
}
static void tsdbEndCompact(STsdbRepo *pRepo, int eno) {
@ -107,7 +109,7 @@ static void tsdbEndCompact(STsdbRepo *pRepo, int eno) {
} else {
tsdbEndFSTxn(pRepo);
}
pRepo->inCompact = false;
pRepo->compactState = TSDB_NO_COMPACT;
tsdbInfo("vgId:%d compact over, %s", REPO_ID(pRepo), (eno == TSDB_CODE_SUCCESS) ? "succeed" : "failed");
tsem_post(&(pRepo->readyToCommit));
}

4
src/tsdb/src/tsdbMain.c
View File

@ -197,7 +197,7 @@ STsdbRepoInfo *tsdbGetStatus(STsdbRepo *pRepo) { return NULL; }
int tsdbGetState(STsdbRepo *repo) { return repo->state; }
bool tsdbInCompact(STsdbRepo *repo) { return repo->inCompact; }
int8_t tsdbGetCompactState(STsdbRepo *repo) { return (int8_t)(repo->compactState); }
void tsdbReportStat(void *repo, int64_t *totalPoints, int64_t *totalStorage, int64_t *compStorage) {
ASSERT(repo != NULL);
@ -537,7 +537,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
pRepo->state = TSDB_STATE_OK;
pRepo->code = TSDB_CODE_SUCCESS;
pRepo->inCompact = false;
pRepo->compactState = 0;
pRepo->config = *pCfg;
if (pAppH) {
pRepo->appH = *pAppH;

2
src/vnode/src/vnodeMgmt.c
View File

@ -160,7 +160,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
pLoad->status = pVnode->status;
pLoad->role = pVnode->role;
pLoad->replica = pVnode->syncCfg.replica;
pLoad->compact = (pVnode->tsdb != NULL) && tsdbInCompact(pVnode->tsdb) ? 1 : 0;
pLoad->compact = (pVnode->tsdb != NULL) ? tsdbGetCompactState(pVnode->tsdb) : 0;
}
int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {