From 055aed43e1389a934feebaeec4aae9bf19890e71 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Mon, 18 May 2020 01:40:08 +0000 Subject: [PATCH] TD-185 --- src/tsdb/inc/tsdbMain.h | 29 +++++--- src/tsdb/src/tsdbFile.c | 38 +--------- src/tsdb/src/tsdbMain.c | 10 ++- src/tsdb/src/tsdbMeta.c | 1 - src/tsdb/src/tsdbRWHelper.c | 136 ++++++++++++++++++++++++++++++++---- src/util/inc/tcoding.h | 10 +++ 6 files changed, 162 insertions(+), 62 deletions(-) diff --git a/src/tsdb/inc/tsdbMain.h b/src/tsdb/inc/tsdbMain.h index 0a1444c2f0..1885b57bc0 100644 --- a/src/tsdb/inc/tsdbMain.h +++ b/src/tsdb/inc/tsdbMain.h @@ -210,12 +210,17 @@ typedef enum { extern const char *tsdbFileSuffix[]; typedef struct { - int64_t size; // total size of the file - int64_t tombSize; // unused file size - int32_t totalBlocks; - int32_t totalSubBlocks; + uint32_t offset; + uint32_t len; + uint64_t size; // total size of the file + uint64_t tombSize; // unused file size + uint32_t totalBlocks; + uint32_t totalSubBlocks; } SFileInfo; +void *tsdbEncodeSFileInfo(void *buf, const SFileInfo *pInfo); +void *tsdbDecodeSFileInfo(void *buf, SFileInfo *pInfo); + typedef struct { int fd; char fname[128]; @@ -242,8 +247,7 @@ typedef struct { STsdbFileH *tsdbInitFileH(char *dataDir, STsdbCfg *pCfg); void tsdbCloseFileH(STsdbFileH *pFileH); -int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, SFile *pFile, int writeHeader, - int toClose); +int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, SFile *pFile); SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables); int tsdbOpenFile(SFile *pFile, int oflag); int tsdbCloseFile(SFile *pFile); @@ -266,15 +270,18 @@ void tsdbSeekFileGroupIter(SFileGroupIter *pIter, int fid); SFileGroup *tsdbGetFileGroupNext(SFileGroupIter *pIter); typedef struct { - int32_t len; - int32_t offset; - int32_t padding; // For padding purpose - int32_t hasLast : 1; - int32_t numOfBlocks : 31; + uint32_t len; + uint32_t offset; + uint32_t padding; // For padding purpose + uint32_t hasLast : 2; + uint32_t numOfBlocks : 30; int64_t uid; TSKEY maxKey; } SCompIdx; /* sizeof(SCompIdx) = 28 */ +void *tsdbEncodeSCompIdx(void *buf, SCompIdx *pIdx); +void *tsdbDecodeSCompIdx(void *buf, SCompIdx *pIdx); + /** * if numOfSubBlocks == 0, then the SCompBlock is a sub-block * if numOfSubBlocks >= 1, then the SCompBlock is a super-block diff --git a/src/tsdb/src/tsdbFile.c b/src/tsdb/src/tsdbFile.c index ad98bb7b20..c9db9344c9 100644 --- a/src/tsdb/src/tsdbFile.c +++ b/src/tsdb/src/tsdbFile.c @@ -38,7 +38,6 @@ const char *tsdbFileSuffix[] = { static int compFGroupKey(const void *key, const void *fgroup); static int compFGroup(const void *arg1, const void *arg2); static int tsdbWriteFileHead(SFile *pFile); -static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables); static int tsdbOpenFGroup(STsdbFileH *pFileH, char *dataDir, int fid); STsdbFileH *tsdbInitFileH(char *dataDir, STsdbCfg *pCfg) { @@ -121,8 +120,7 @@ SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int max if (pGroup == NULL) { // if not exists, create one pFGroup->fileId = fid; for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) { - if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], maxTables, &(pFGroup->files[type]), - type == TSDB_FILE_TYPE_HEAD ? 1 : 0, 1) < 0) + if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], &(pFGroup->files[type])) < 0) goto _err; } @@ -298,29 +296,6 @@ static int tsdbWriteFileHead(SFile *pFile) { return 0; } -static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables) { - int size = sizeof(SCompIdx) * maxTables + sizeof(TSCKSUM); - void *buf = calloc(1, size); - if (buf == NULL) return -1; - - if (lseek(pFile->fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) { - free(buf); - return -1; - } - - taosCalcChecksumAppend(0, (uint8_t *)buf, size); - - if (write(pFile->fd, buf, size) < 0) { - free(buf); - return -1; - } - - pFile->info.size += size; - - free(buf); - return 0; -} - int tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname) { if (dataDir == NULL || fname == NULL) return -1; @@ -354,7 +329,7 @@ SFileGroup * tsdbOpenFilesForCommit(STsdbFileH *pFileH, int fid) { return pGroup; } -int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, SFile *pFile, int writeHeader, int toClose) { +int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, SFile *pFile) { memset((void *)pFile, 0, sizeof(SFile)); pFile->fd = -1; @@ -370,19 +345,12 @@ int tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, return -1; } - if (writeHeader) { - if (tsdbWriteHeadFileIdx(pFile, maxTables) < 0) { - tsdbCloseFile(pFile); - return -1; - } - } - if (tsdbWriteFileHead(pFile) < 0) { tsdbCloseFile(pFile); return -1; } - if (toClose) tsdbCloseFile(pFile); + tsdbCloseFile(pFile); return 0; } diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c index c63de165d3..7254318d74 100644 --- a/src/tsdb/src/tsdbMain.c +++ b/src/tsdb/src/tsdbMain.c @@ -1148,8 +1148,16 @@ static void tsdbAlterKeep(STsdbRepo *pRepo, int32_t keep) { } static void tsdbAlterMaxTables(STsdbRepo *pRepo, int32_t maxTables) { - // TODO int oldMaxTables = pRepo->config.maxTables; + if (oldMaxTables < pRepo->config.maxTables) { + // TODO + } + + STsdbMeta *pMeta = pRepo->tsdbMeta; + + pMeta->maxTables = maxTables; + pMeta->tables = realloc(pMeta->tables, maxTables * sizeof(STable *)); + tsdbTrace("vgId:%d, tsdb maxTables is changed from %d to %d!", pRepo->config.tsdbId, oldMaxTables, maxTables); } diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c index aaea50267d..1b161a4ad7 100644 --- a/src/tsdb/src/tsdbMeta.c +++ b/src/tsdb/src/tsdbMeta.c @@ -154,7 +154,6 @@ STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables) { STsdbMeta *pMeta = (STsdbMeta *)malloc(sizeof(STsdbMeta)); if (pMeta == NULL) return NULL; - pMeta->maxTables = maxTables; pMeta->nTables = 0; pMeta->superList = NULL; pMeta->tables = (STable **)calloc(maxTables, sizeof(STable *)); diff --git a/src/tsdb/src/tsdbRWHelper.c b/src/tsdb/src/tsdbRWHelper.c index 829d7ab3fd..350fde412c 100644 --- a/src/tsdb/src/tsdbRWHelper.c +++ b/src/tsdb/src/tsdbRWHelper.c @@ -18,6 +18,7 @@ #include "tchecksum.h" #include "tscompression.h" #include "talgo.h" +#include "tcoding.h" // Local function definitions // static int tsdbCheckHelperCfg(SHelperCfg *pCfg); @@ -211,13 +212,15 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) { // Create and open .h if (tsdbOpenFile(&(pHelper->files.nHeadF), O_WRONLY | O_CREAT) < 0) return -1; - size_t tsize = TSDB_FILE_HEAD_SIZE + sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM); - if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, tsize) < tsize) goto _err; + // size_t tsize = TSDB_FILE_HEAD_SIZE + sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM); + if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) + goto _err; // Create and open .l file if should if (tsdbShouldCreateNewLast(pHelper)) { if (tsdbOpenFile(&(pHelper->files.nLastF), O_WRONLY | O_CREAT) < 0) goto _err; - if (tsendfile(pHelper->files.nLastF.fd, pHelper->files.lastF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) goto _err; + if (tsendfile(pHelper->files.nLastF.fd, pHelper->files.lastF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) + goto _err; } } else { if (tsdbOpenFile(&(pHelper->files.dataF), O_RDONLY) < 0) goto _err; @@ -426,13 +429,26 @@ int tsdbWriteCompInfo(SRWHelper *pHelper) { int tsdbWriteCompIdx(SRWHelper *pHelper) { ASSERT(TSDB_HELPER_TYPE(pHelper) == TSDB_WRITE_HELPER); - if (lseek(pHelper->files.nHeadF.fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1; + off_t offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END); + if (offset < 0) return -1; + + SFile *pFile = &(pHelper->files.nHeadF); + pFile->info.offset = offset; + + void *buf = pHelper->blockBuffer; + for (uint32_t i = 0; i < pHelper->config.maxTables; i++) { + SCompIdx *pCompIdx = pHelper->pCompIdx + i; + if (pCompIdx->offset > 0) { + buf = taosEncodeVariant32(buf, i); + buf = tsdbEncodeSCompIdx(buf, pCompIdx); + } + } - ASSERT(tsizeof(pHelper->pCompIdx) == sizeof(SCompIdx) * pHelper->config.maxTables + sizeof(TSCKSUM)); - taosCalcChecksumAppend(0, (uint8_t *)pHelper->pCompIdx, tsizeof(pHelper->pCompIdx)); + int tsize = (char *)buf - (char *)pHelper->blockBuffer + sizeof(TSCKSUM); + taosCalcChecksumAppend(0, (uint8_t *)pHelper->blockBuffer, tsize); - if (twrite(pHelper->files.nHeadF.fd, (void *)pHelper->pCompIdx, tsizeof(pHelper->pCompIdx)) < tsizeof(pHelper->pCompIdx)) - return -1; + if (twrite(pHelper->files.nHeadF.fd, (void *)pHelper->blockBuffer, tsize) < tsize) return -1; + pFile->info.len = tsize; return 0; } @@ -441,14 +457,34 @@ int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) { if (!helperHasState(pHelper, TSDB_HELPER_IDX_LOAD)) { // If not load from file, just load it in object - int fd = pHelper->files.headF.fd; + SFile *pFile = &(pHelper->files.headF); + int fd = pFile->fd; + + memset(pHelper->pCompIdx, 0, tsizeof(pHelper->pCompIdx)); + if (pFile->info.offset > 0) { + ASSERT(pFile->info.offset > TSDB_FILE_HEAD_SIZE); + + if (lseek(fd, pFile->info.offset, SEEK_SET) < 0) return -1; + if (tread(fd, (void *)(pHelper->blockBuffer), pFile->info.len) < pFile->info.len) + return -1; + if (!taosCheckChecksumWhole((uint8_t *)(pHelper->blockBuffer), pFile->info.len)) { + // TODO: File is broken, try to deal with it + return -1; + } - if (lseek(fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1; - if (tread(fd, (void *)(pHelper->pCompIdx), tsizeof((void *)pHelper->pCompIdx)) < tsizeof(pHelper->pCompIdx)) return -1; - if (!taosCheckChecksumWhole((uint8_t *)(pHelper->pCompIdx), tsizeof((void *)pHelper->pCompIdx))) { - // TODO: File is broken, try to deal with it - return -1; + // Decode it + void *ptr = pHelper->blockBuffer; + while ((char *)ptr - (char *)pHelper->blockBuffer >= pFile->info.len - sizeof(TSCKSUM)) { + uint32_t tid = 0; + if ((ptr = taosDecodeVariant32(ptr, &tid)) == NULL) return -1; + ASSERT(tid > 0 && tid < pHelper->config.maxTables); + + if ((ptr = tsdbDecodeSCompIdx(ptr, pHelper->pCompIdx + tid)) == NULL) return -1; + + ASSERT((char *)ptr - (char *)pHelper->blockBuffer <= pFile->info.len - sizeof(TSCKSUM)); + } } + } helperSetState(pHelper, TSDB_HELPER_IDX_LOAD); @@ -1168,4 +1204,76 @@ static int tsdbGetRowsInRange(SDataCols *pDataCols, TSKEY minKey, TSKEY maxKey) if ((TSKEY *)ptr2 - (TSKEY *)ptr1 < 0) return 0; return ((TSKEY *)ptr2 - (TSKEY *)ptr1) + 1; +} + +void *tsdbEncodeSCompIdx(void *buf, SCompIdx *pIdx) { + buf = taosEncodeVariant32(buf, pIdx->len); + buf = taosEncodeVariant32(buf, pIdx->offset); + buf = taosEncodeFixed8(buf, pIdx->hasLast); + buf = taosEncodeVariant32(buf, pIdx->numOfBlocks); + buf = taosEncodeFixed64(buf, pIdx->uid); + buf = taosEncodeFixed64(buf, pIdx->maxKey); + + return buf; +} + +void *tsdbDecodeSCompIdx(void *buf, SCompIdx *pIdx) { + uint8_t hasLast = 0; + uint32_t numOfBlocks = 0; + uint64_t value = 0; + + if ((buf = taosDecodeVariant32(buf, &(pIdx->len))) == NULL) return NULL; + if ((buf = taosDecodeVariant32(buf, &(pIdx->offset))) == NULL) return NULL; + if ((buf = taosDecodeFixed8(buf, &(hasLast))) == NULL) return NULL; + pIdx->hasLast = hasLast; + if ((buf = taosDecodeVariant32(buf, &(numOfBlocks))) == NULL) return NULL; + pIdx->numOfBlocks = numOfBlocks; + if ((buf = taosDecodeFixed64(buf, &value)) == NULL) return NULL; + pIdx->uid = (int64_t)value; + if ((buf = taosDecodeFixed64(buf, &value)) == NULL) return NULL; + pIdx->maxKey = (TSKEY)value; + + return buf; +} + +int tsdbUpdateFileHeader(SFile *pFile, uint32_t version) { + char buf[TSDB_FILE_HEAD_SIZE] = "\0"; + + void *pBuf = (void *)buf; + pBuf = taosEncodeFixed32(pBuf, version); + pBuf = tsdbEncodeSFileInfo(pBuf, &(pFile->info)); + int tsize = (char *)pBuf - buf + sizeof(TSCKSUM); + + ASSERT(tsize <= TSDB_FILE_HEAD_SIZE); + + taosCalcChecksumAppend(0, (uint8_t *)buf, tsize); + + if (lseek(pFile->fd, 0, SEEK_SET) < 0) return -1; + if (twrite(pFile->fd, (void *)buf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) return -1; + + return 0; +} + + + +void *tsdbEncodeSFileInfo(void *buf, const SFileInfo *pInfo) { + buf = taosEncodeFixed32(buf, pInfo->offset); + buf = taosEncodeFixed32(buf, pInfo->len); + buf = taosEncodeFixed64(buf, pInfo->size); + buf = taosEncodeFixed64(buf, pInfo->tombSize); + buf = taosEncodeFixed32(buf, pInfo->totalBlocks); + buf = taosEncodeFixed32(buf, pInfo->totalSubBlocks); + + return buf; +} + +void *tsdbDecodeSFileInfo(void *buf, SFileInfo *pInfo) { + buf = taosDecodeFixed32(buf, &(pInfo->offset)); + buf = taosDecodeFixed32(buf, &(pInfo->len)); + buf = taosDecodeFixed64(buf, &(pInfo->size)); + buf = taosDecodeFixed64(buf, &(pInfo->tombSize)); + buf = taosDecodeFixed32(buf, &(pInfo->totalBlocks)); + buf = taosDecodeFixed32(buf, &(pInfo->totalSubBlocks)); + + return buf; } \ No newline at end of file diff --git a/src/util/inc/tcoding.h b/src/util/inc/tcoding.h index 9f64f127e1..b4f7f596c5 100644 --- a/src/util/inc/tcoding.h +++ b/src/util/inc/tcoding.h @@ -29,6 +29,11 @@ extern "C" { static const int32_t TNUMBER = 1; #define IS_LITTLE_ENDIAN() (*(uint8_t *)(&TNUMBER) != 0) +static FORCE_INLINE void *taosEncodeFixed8(void *buf, uint8_t value) { + ((uint8_t *)buf)[0] = value; + return POINTER_SHIFT(buf, sizeof(value)); +} + static FORCE_INLINE void *taosEncodeFixed16(void *buf, uint16_t value) { if (IS_LITTLE_ENDIAN()) { memcpy(buf, &value, sizeof(value)); @@ -70,6 +75,11 @@ static FORCE_INLINE void *taosEncodeFixed64(void *buf, uint64_t value) { return POINTER_SHIFT(buf, sizeof(value)); } +static FORCE_INLINE void *taosDecodeFixed8(void *buf, uint8_t *value) { + *value = ((uint8_t *)buf)[0]; + return POINTER_SHIFT(buf, sizeof(*value)); +} + static FORCE_INLINE void *taosDecodeFixed16(void *buf, uint16_t *value) { if (IS_LITTLE_ENDIAN()) { memcpy(value, buf, sizeof(*value)); -- GitLab