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

tsdb r/w concurrency

This commit is contained in:
Hongze Cheng 2022-07-19 06:19:01 +00:00
parent 007c5eb0a5
commit 58135a1aeb
8 changed files with 137 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
source/dnode/vnode/src/inc/tsdb.h
View File

@ -174,6 +174,10 @@ void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg);
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
void tsdbMemTableDestroy(SMemTable *pMemTable);
void tsdbGetTbDataFromMemTable(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData);
void tsdbRefMemTable(SMemTable *pMemTable);
void tsdbUnrefMemTable(SMemTable *pMemTable);
int32_t tsdbTakeMemSnapshot(STsdb *pTsdb, SMemTable **ppMem, SMemTable **ppIMem);
void tsdbUntakeMemSnapshot(STsdb *pTsdb, SMemTable *pMem, SMemTable *pIMem);
// STbDataIter
int32_t tsdbTbDataIterCreate(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter **ppIter);
void *tsdbTbDataIterDestroy(STbDataIter *pIter);
@ -273,16 +277,14 @@ typedef struct {
} SRtn;
struct STsdb {
char *path;
SVnode *pVnode;
TdThreadMutex mutex;
bool repoLocked;
STsdbKeepCfg keepCfg;
SMemTable *mem;
SMemTable *imem;
SRtn rtn;
STsdbFS *fs;
SLRUCache *lruCache;
char *path;
SVnode *pVnode;
TdThreadRwlock rwLock;
STsdbKeepCfg keepCfg;
SMemTable *mem;
SMemTable *imem;
STsdbFS *pFS;
SLRUCache *lruCache;
};
struct STable {
@ -330,21 +332,19 @@ struct STbData {
};
struct SMemTable {
SRWLatch latch;
STsdb *pTsdb;
int32_t nRef;
TSKEY minKey;
TSKEY maxKey;
int64_t minVersion;
int64_t maxVersion;
int64_t nRow;
int64_t nDel;
SArray *aTbData; // SArray<STbData*>
SRWLatch latch;
STsdb *pTsdb;
SVBufPool *pPool;
volatile int32_t nRef;
TSKEY minKey;
TSKEY maxKey;
int64_t minVersion;
int64_t maxVersion;
int64_t nRow;
int64_t nDel;
SArray *aTbData; // SArray<STbData*>
};
int tsdbLockRepo(STsdb *pTsdb);
int tsdbUnlockRepo(STsdb *pTsdb);
struct TSDBROW {
int8_t type; // 0 for row from tsRow, 1 for row from block data
union {

8
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -464,7 +464,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
switch (state->state) {
case SFSNEXTROW_FS:
state->aDFileSet = state->pTsdb->fs->cState->aDFileSet;
state->aDFileSet = state->pTsdb->pFS->cState->aDFileSet;
state->nFileSet = taosArrayGetSize(state->aDFileSet);
state->iFileSet = state->nFileSet;
@ -814,7 +814,7 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
SDelIdx delIdx;
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->fs->cState);
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState);
if (pDelFile) {
SDelFReader *pDelFReader;
@ -1189,7 +1189,7 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
SDelIdx delIdx;
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->fs->cState);
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState);
if (pDelFile) {
SDelFReader *pDelFReader;
@ -1377,7 +1377,7 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
SDelIdx delIdx;
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->fs->cState);
SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState);
if (pDelFile) {
SDelFReader *pDelFReader;

44
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -64,9 +64,26 @@ int32_t tsdbBegin(STsdb *pTsdb) {
if (!pTsdb) return code;
code = tsdbMemTableCreate(pTsdb, &pTsdb->mem);
SMemTable *pMemTable;
code = tsdbMemTableCreate(pTsdb, &pMemTable);
if (code) goto _err;
// lock
code = taosThreadRwlockWrlock(&pTsdb->rwLock);
if (code) {
code = TAOS_SYSTEM_ERROR(code);
goto _err;
}
pTsdb->mem = pMemTable;
// unlock
code = taosThreadRwlockUnlock(&pTsdb->rwLock);
if (code) {
code = TAOS_SYSTEM_ERROR(code);
goto _err;
}
return code;
_err:
@ -83,9 +100,11 @@ int32_t tsdbCommit(STsdb *pTsdb) {
// check
if (pMemTable->nRow == 0 && pMemTable->nDel == 0) {
// TODO: lock?
taosThreadRwlockWrlock(&pTsdb->rwLock);
pTsdb->mem = NULL;
tsdbMemTableDestroy(pMemTable);
taosThreadRwlockUnlock(&pTsdb->rwLock);
tsdbUnrefMemTable(pMemTable);
goto _exit;
}
@ -139,7 +158,7 @@ static int32_t tsdbCommitDelStart(SCommitter *pCommitter) {
goto _err;
}
SDelFile *pDelFileR = pTsdb->fs->nState->pDelFile;
SDelFile *pDelFileR = pTsdb->pFS->nState->pDelFile;
if (pDelFileR) {
code = tsdbDelFReaderOpen(&pCommitter->pDelFReader, pDelFileR, pTsdb, NULL);
if (code) goto _err;
@ -228,7 +247,7 @@ static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) {
code = tsdbUpdateDelFileHdr(pCommitter->pDelFWriter);
if (code) goto _err;
code = tsdbFSStateUpsertDelFile(pTsdb->fs->nState, &pCommitter->pDelFWriter->fDel);
code = tsdbFSStateUpsertDelFile(pTsdb->pFS->nState, &pCommitter->pDelFWriter->fDel);
if (code) goto _err;
code = tsdbDelFWriterClose(&pCommitter->pDelFWriter, 1);
@ -263,7 +282,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
taosArrayClear(pCommitter->aBlockIdx);
tMapDataReset(&pCommitter->oBlockMap);
tBlockDataReset(&pCommitter->oBlockData);
pRSet = tsdbFSStateGetDFileSet(pTsdb->fs->nState, pCommitter->commitFid, TD_EQ);
pRSet = tsdbFSStateGetDFileSet(pTsdb->pFS->nState, pCommitter->commitFid, TD_EQ);
if (pRSet) {
code = tsdbDataFReaderOpen(&pCommitter->pReader, pTsdb, pRSet);
if (code) goto _err;
@ -836,7 +855,7 @@ static int32_t tsdbCommitFileDataEnd(SCommitter *pCommitter) {
if (code) goto _err;
// upsert SDFileSet
code = tsdbFSStateUpsertDFileSet(pCommitter->pTsdb->fs->nState, tsdbDataFWriterGetWSet(pCommitter->pWriter));
code = tsdbFSStateUpsertDFileSet(pCommitter->pTsdb->pFS->nState, tsdbDataFWriterGetWSet(pCommitter->pWriter));
if (code) goto _err;
// close and sync
@ -954,7 +973,7 @@ static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) {
pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
code = tsdbFSBegin(pTsdb->fs);
code = tsdbFSBegin(pTsdb->pFS);
if (code) goto _err;
return code;
@ -1135,13 +1154,16 @@ static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno) {
SMemTable *pMemTable = pTsdb->imem;
if (eno == 0) {
code = tsdbFSCommit(pTsdb->fs);
code = tsdbFSCommit(pTsdb->pFS);
} else {
code = tsdbFSRollback(pTsdb->fs);
code = tsdbFSRollback(pTsdb->pFS);
}
tsdbMemTableDestroy(pMemTable);
taosThreadRwlockWrlock(&pTsdb->rwLock);
pTsdb->imem = NULL;
taosThreadRwlockUnlock(&pTsdb->rwLock);
tsdbUnrefMemTable(pMemTable);
tsdbInfo("vgId:%d tsdb end commit", TD_VID(pTsdb->pVnode));
return code;

56
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -41,6 +41,7 @@ int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
}
taosInitRWLatch(&pMemTable->latch);
pMemTable->pTsdb = pTsdb;
pMemTable->pPool = pTsdb->pVnode->inUse;
pMemTable->nRef = 1;
pMemTable->minKey = TSKEY_MAX;
pMemTable->maxKey = TSKEY_MIN;
@ -590,3 +591,58 @@ _err:
}
int32_t tsdbGetNRowsInTbData(STbData *pTbData) { return pTbData->sl.size; }
void tsdbRefMemTable(SMemTable *pMemTable) {
int32_t nRef = atomic_fetch_add_32(&pMemTable->nRef, 1);
ASSERT(nRef > 0);
}
void tsdbUnrefMemTable(SMemTable *pMemTable) {
int32_t nRef = atomic_sub_fetch_32(&pMemTable->nRef, 1);
if (nRef == 0) {
tsdbMemTableDestroy(pMemTable);
}
}
int32_t tsdbTakeMemSnapshot(STsdb *pTsdb, SMemTable **ppMem, SMemTable **ppIMem) {
int32_t code = 0;
// lock
code = taosThreadRwlockRdlock(&pTsdb->rwLock);
if (code) {
code = TAOS_SYSTEM_ERROR(code);
goto _exit;
}
// take snapshot
*ppMem = pTsdb->mem;
*ppIMem = pTsdb->imem;
if (*ppMem) {
tsdbRefMemTable(*ppMem);
}
if (*ppIMem) {
tsdbRefMemTable(*ppIMem);
}
// unlock
code = taosThreadRwlockUnlock(&pTsdb->rwLock);
if (code) {
code = TAOS_SYSTEM_ERROR(code);
goto _exit;
}
_exit:
return code;
}
void tsdbUntakeMemSnapshot(STsdb *pTsdb, SMemTable *pMem, SMemTable *pIMem) {
if (pMem) {
tsdbUnrefMemTable(pMem);
}
if (pIMem) {
tsdbUnrefMemTable(pIMem);
}
}

32
source/dnode/vnode/src/tsdb/tsdbOpen.c
View File

@ -54,8 +54,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
sprintf(pTsdb->path, "%s%s%s", pVnode->path, TD_DIRSEP, dir);
taosRealPath(pTsdb->path, NULL, slen);
pTsdb->pVnode = pVnode;
pTsdb->repoLocked = false;
taosThreadMutexInit(&pTsdb->mutex, NULL);
taosThreadRwlockInit(&pTsdb->rwLock, NULL);
if (!pKeepCfg) {
tsdbSetKeepCfg(&pTsdb->keepCfg, &pVnode->config.tsdbCfg);
} else {
@ -67,7 +66,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
tfsMkdir(pVnode->pTfs, pTsdb->path);
// open tsdb
if (tsdbFSOpen(pTsdb, &pTsdb->fs) < 0) {
if (tsdbFSOpen(pTsdb, &pTsdb->pFS) < 0) {
goto _err;
}
@ -88,33 +87,10 @@ _err:
int tsdbClose(STsdb **pTsdb) {
if (*pTsdb) {
taosThreadMutexDestroy(&(*pTsdb)->mutex);
tsdbFSClose((*pTsdb)->fs);
taosThreadRwlockDestroy(&(*pTsdb)->rwLock);
tsdbFSClose((*pTsdb)->pFS);
tsdbCloseCache((*pTsdb)->lruCache);
taosMemoryFreeClear(*pTsdb);
}
return 0;
}
int tsdbLockRepo(STsdb *pTsdb) {
int code = taosThreadMutexLock(&pTsdb->mutex);
if (code != 0) {
tsdbError("vgId:%d, failed to lock tsdb since %s", TD_VID(pTsdb->pVnode), strerror(errno));
terrno = TAOS_SYSTEM_ERROR(code);
return -1;
}
pTsdb->repoLocked = true;
return 0;
}
int tsdbUnlockRepo(STsdb *pTsdb) {
// ASSERT(IS_REPO_LOCKED(pTsdb));
pTsdb->repoLocked = false;
int code = taosThreadMutexUnlock(&pTsdb->mutex);
if (code != 0) {
tsdbError("vgId:%d, failed to unlock tsdb since %s", TD_VID(pTsdb->pVnode), strerror(errno));
terrno = TAOS_SYSTEM_ERROR(code);
return -1;
}
return 0;
}

6
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -1905,7 +1905,7 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader*
SArray* pDelData = taosArrayInit(4, sizeof(SDelData));
SDelFile* pDelFile = tsdbFSStateGetDelFile(pTsdb->fs->cState);
SDelFile* pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState);
if (pDelFile) {
SDelFReader* pDelFReader = NULL;
code = tsdbDelFReaderOpen(&pDelFReader, pDelFile, pTsdb, NULL);
@ -2795,7 +2795,7 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTabl
SDataBlockIter* pBlockIter = &pReader->status.blockIter;
STsdbFSState* pFState = pReader->pTsdb->fs->cState;
STsdbFSState* pFState = pReader->pTsdb->pFS->cState;
initFilesetIterator(&pReader->status.fileIter, pFState, pReader->order, pReader->idStr);
resetDataBlockIterator(&pReader->status.blockIter, pReader->order);
@ -3042,7 +3042,7 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) {
tsdbDataFReaderClose(&pReader->pFileReader);
STsdbFSState* pFState = pReader->pTsdb->fs->cState;
STsdbFSState* pFState = pReader->pTsdb->pFS->cState;
initFilesetIterator(&pReader->status.fileIter, pFState, pReader->order, pReader->idStr);
resetDataBlockIterator(&pReader->status.blockIter, pReader->order);
resetDataBlockScanInfo(pReader->status.pTableMap);

10
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -20,10 +20,10 @@ static int32_t tsdbDoRetentionImpl(STsdb *pTsdb, int64_t now, int8_t try, int8_t
STsdbFSState *pState;
if (try) {
pState = pTsdb->fs->cState;
pState = pTsdb->pFS->cState;
*canDo = 0;
} else {
pState = pTsdb->fs->nState;
pState = pTsdb->pFS->nState;
}
for (int32_t iSet = 0; iSet < taosArrayGetSize(pState->aDFileSet); iSet++) {
@ -83,7 +83,7 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
if (!canDo) goto _exit;
// begin
code = tsdbFSBegin(pTsdb->fs);
code = tsdbFSBegin(pTsdb->pFS);
if (code) goto _err;
// do retention
@ -91,7 +91,7 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
if (code) goto _err;
// commit
code = tsdbFSCommit(pTsdb->fs);
code = tsdbFSCommit(pTsdb->pFS);
if (code) goto _err;
_exit:
@ -99,6 +99,6 @@ _exit:
_err:
tsdbError("vgId:%d tsdb do retention failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
tsdbFSRollback(pTsdb->fs);
tsdbFSRollback(pTsdb->pFS);
return code;
}

18
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -45,7 +45,7 @@ static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
while (true) {
if (pReader->pDataFReader == NULL) {
SDFileSet* pSet = tsdbFSStateGetDFileSet(pTsdb->fs->cState, pReader->fid, TD_GT);
SDFileSet* pSet = tsdbFSStateGetDFileSet(pTsdb->pFS->cState, pReader->fid, TD_GT);
if (pSet == NULL) goto _exit;
@ -159,7 +159,7 @@ _err:
static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0;
STsdb* pTsdb = pReader->pTsdb;
SDelFile* pDelFile = pTsdb->fs->cState->pDelFile;
SDelFile* pDelFile = pTsdb->pFS->cState->pDelFile;
if (pReader->pDelFReader == NULL) {
if (pDelFile == NULL) {
@ -798,7 +798,7 @@ static int32_t tsdbSnapWriteDataEnd(STsdbSnapWriter* pWriter) {
code = tsdbWriteBlockIdx(pWriter->pDataFWriter, pWriter->aBlockIdxW, NULL);
if (code) goto _err;
code = tsdbFSStateUpsertDFileSet(pTsdb->fs->nState, tsdbDataFWriterGetWSet(pWriter->pDataFWriter));
code = tsdbFSStateUpsertDFileSet(pTsdb->pFS->nState, tsdbDataFWriterGetWSet(pWriter->pDataFWriter));
if (code) goto _err;
code = tsdbDataFWriterClose(&pWriter->pDataFWriter, 1);
@ -843,7 +843,7 @@ static int32_t tsdbSnapWriteData(STsdbSnapWriter* pWriter, uint8_t* pData, uint3
pWriter->fid = fid;
// read
SDFileSet* pSet = tsdbFSStateGetDFileSet(pTsdb->fs->nState, fid, TD_EQ);
SDFileSet* pSet = tsdbFSStateGetDFileSet(pTsdb->pFS->nState, fid, TD_EQ);
if (pSet) {
code = tsdbDataFReaderOpen(&pWriter->pDataFReader, pTsdb, pSet);
if (code) goto _err;
@ -907,7 +907,7 @@ static int32_t tsdbSnapWriteDel(STsdbSnapWriter* pWriter, uint8_t* pData, uint32
STsdb* pTsdb = pWriter->pTsdb;
if (pWriter->pDelFWriter == NULL) {
SDelFile* pDelFile = tsdbFSStateGetDelFile(pTsdb->fs->nState);
SDelFile* pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->nState);
// reader
if (pDelFile) {
@ -1017,7 +1017,7 @@ static int32_t tsdbSnapWriteDelEnd(STsdbSnapWriter* pWriter) {
code = tsdbUpdateDelFileHdr(pWriter->pDelFWriter);
if (code) goto _err;
code = tsdbFSStateUpsertDelFile(pTsdb->fs->nState, &pWriter->pDelFWriter->fDel);
code = tsdbFSStateUpsertDelFile(pTsdb->pFS->nState, &pWriter->pDelFWriter->fDel);
if (code) goto _err;
code = tsdbDelFWriterClose(&pWriter->pDelFWriter, 1);
@ -1096,7 +1096,7 @@ int32_t tsdbSnapWriterOpen(STsdb* pTsdb, int64_t sver, int64_t ever, STsdbSnapWr
goto _err;
}
code = tsdbFSBegin(pTsdb->fs);
code = tsdbFSBegin(pTsdb->pFS);
if (code) goto _err;
*ppWriter = pWriter;
@ -1113,7 +1113,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
STsdbSnapWriter* pWriter = *ppWriter;
if (rollback) {
code = tsdbFSRollback(pWriter->pTsdb->fs);
code = tsdbFSRollback(pWriter->pTsdb->pFS);
if (code) goto _err;
} else {
code = tsdbSnapWriteDataEnd(pWriter);
@ -1122,7 +1122,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
code = tsdbSnapWriteDelEnd(pWriter);
if (code) goto _err;
code = tsdbFSCommit(pWriter->pTsdb->fs);
code = tsdbFSCommit(pWriter->pTsdb->pFS);
if (code) goto _err;
}