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

[TD-3963]sync config change msg and wait for the config applied

This commit is contained in:
lichuang 2021-06-04 14:50:47 +08:00
parent 864b0d2792
commit d7219d1aaa
3 changed files with 13 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/tsdb/inc/tsdbMemTable.h
View File

@ -66,7 +66,7 @@ int tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemSnapshot* pSnapshot, SArray* pAT
void tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemSnapshot* pSnapshot); void tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemSnapshot* pSnapshot);
void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes); void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
int tsdbAsyncCommit(STsdbRepo* pRepo); int tsdbAsyncCommit(STsdbRepo* pRepo);
int tsdbAsyncCommitConfig(STsdbRepo* pRepo); int tsdbSyncCommitConfig(STsdbRepo* pRepo);
int tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols, int tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,
TKEY* filterKeys, int nFilterKeys, bool keepDup, SMergeInfo* pMergeInfo); TKEY* filterKeys, int nFilterKeys, bool keepDup, SMergeInfo* pMergeInfo);
void* tsdbCommitData(STsdbRepo* pRepo); void* tsdbCommitData(STsdbRepo* pRepo);

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

@ -270,8 +270,8 @@ int32_t tsdbConfigRepo(STsdbRepo *repo, STsdbCfg *pCfg) {
pthread_mutex_unlock(&repo->save_mutex); pthread_mutex_unlock(&repo->save_mutex);
// schedule a commit msg then the new config will be applied immediatly // schedule a commit msg and wait for the new config applied
tsdbAsyncCommitConfig(repo); tsdbSyncCommitConfig(repo);
return 0; return 0;
#if 0 #if 0

11
src/tsdb/src/tsdbMemTable.c
View File

@ -271,7 +271,7 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
return ptr; return ptr;
} }
int tsdbAsyncCommitConfig(STsdbRepo* pRepo) { int tsdbSyncCommitConfig(STsdbRepo* pRepo) {
ASSERT(pRepo->config_changed == true); ASSERT(pRepo->config_changed == true);
tsem_wait(&(pRepo->readyToCommit)); tsem_wait(&(pRepo->readyToCommit));
@ -283,6 +283,15 @@ int tsdbAsyncCommitConfig(STsdbRepo* pRepo) {
tsdbScheduleCommit(pRepo, COMMIT_CONFIG_REQ); tsdbScheduleCommit(pRepo, COMMIT_CONFIG_REQ);
if (tsdbUnlockRepo(pRepo) < 0) return -1; if (tsdbUnlockRepo(pRepo) < 0) return -1;
tsem_wait(&(pRepo->readyToCommit));
tsem_post(&(pRepo->readyToCommit));
if (pRepo->code != TSDB_CODE_SUCCESS) {
terrno = pRepo->code;
return -1;
}
terrno = TSDB_CODE_SUCCESS;
return 0; return 0;
} }