From e50299ce987d9809c9730458ee22d03282d88e6a Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Wed, 3 Nov 2021 18:48:30 +0800 Subject: [PATCH 1/3] add unit test for tqMeta --- source/dnode/vnode/tq/CMakeLists.txt | 4 + source/dnode/vnode/tq/inc/tqMetaStore.h | 31 ++-- source/dnode/vnode/tq/src/tqMetaStore.c | 169 ++++++++++++++-------- source/dnode/vnode/tq/test/CMakeLists.txt | 20 +++ source/dnode/vnode/tq/test/tqMetaTest.cpp | 97 +++++++++++++ source/dnode/vnode/tq/test/tqTests.cpp | 0 source/os/src/osDir.c | 4 +- 7 files changed, 248 insertions(+), 77 deletions(-) create mode 100644 source/dnode/vnode/tq/test/CMakeLists.txt create mode 100644 source/dnode/vnode/tq/test/tqMetaTest.cpp delete mode 100644 source/dnode/vnode/tq/test/tqTests.cpp diff --git a/source/dnode/vnode/tq/CMakeLists.txt b/source/dnode/vnode/tq/CMakeLists.txt index 441fe46244..f9e86adc73 100644 --- a/source/dnode/vnode/tq/CMakeLists.txt +++ b/source/dnode/vnode/tq/CMakeLists.txt @@ -12,3 +12,7 @@ target_link_libraries( PUBLIC os PUBLIC util ) + +if(${BUILD_TEST}) + add_subdirectory(test) +endif(${BUILD_TEST}) diff --git a/source/dnode/vnode/tq/inc/tqMetaStore.h b/source/dnode/vnode/tq/inc/tqMetaStore.h index 65429f7af2..f3918d0a8e 100644 --- a/source/dnode/vnode/tq/inc/tqMetaStore.h +++ b/source/dnode/vnode/tq/inc/tqMetaStore.h @@ -19,6 +19,11 @@ #include "os.h" #include "tq.h" + +#ifdef __cplusplus +extern "C" { +#endif + #define TQ_BUCKET_SIZE 0xFF #define TQ_PAGE_SIZE 4096 //key + offset + size @@ -32,10 +37,6 @@ inline static int TqEmptyTail() { //16 return TQ_PAGE_SIZE - TqMaxEntryOnePage(); } -#ifdef __cplusplus -extern "C" { -#endif - typedef struct TqMetaHandle { int64_t key; int64_t offset; @@ -59,30 +60,30 @@ typedef struct TqMetaStore { TqMetaList* unpersistHead; int fileFd; //TODO:temporaral use, to be replaced by unified tfile int idxFd; //TODO:temporaral use, to be replaced by unified tfile - int (*serializer)(TqGroupHandle*, void**); - const void* (*deserializer)(const void*, TqGroupHandle*); + int (*serializer)(const void* pObj, void** ppBytes); + const void* (*deserializer)(const void* pBytes, void** ppObj); void (*deleter)(void*); } TqMetaStore; TqMetaStore* tqStoreOpen(const char* path, - int serializer(TqGroupHandle*, void**), - const void* deserializer(const void*, TqGroupHandle*), - void deleter(void*)); + int serializer(const void* pObj, void** ppBytes), + const void* deserializer(const void* pBytes, void** ppObj), + void deleter(void* pObj)); int32_t tqStoreClose(TqMetaStore*); //int32_t tqStoreDelete(TqMetaStore*); //int32_t TqStoreCommitAll(TqMetaStore*); int32_t tqStorePersist(TqMetaStore*); -TqMetaHandle* tqHandleGet(TqMetaStore*, int64_t key); -int32_t tqHandlePut(TqMetaStore*, int64_t key, void* value); +void* tqHandleGet(TqMetaStore*, int64_t key); +int32_t tqHandlePut(TqMetaStore*, int64_t key, void* value); //do commit -int32_t tqHandleCommit(TqMetaStore*, int64_t key); +int32_t tqHandleCommit(TqMetaStore*, int64_t key); //delete uncommitted -int32_t tqHandleAbort(TqMetaStore*, int64_t key); +int32_t tqHandleAbort(TqMetaStore*, int64_t key); //delete committed -int32_t tqHandleDel(TqMetaStore*, int64_t key); +int32_t tqHandleDel(TqMetaStore*, int64_t key); //delete both committed and uncommitted -int32_t tqHandleClear(TqMetaStore*, int64_t key); +int32_t tqHandleClear(TqMetaStore*, int64_t key); #ifdef __cplusplus } diff --git a/source/dnode/vnode/tq/src/tqMetaStore.c b/source/dnode/vnode/tq/src/tqMetaStore.c index 72edf10a1f..f850f73e12 100644 --- a/source/dnode/vnode/tq/src/tqMetaStore.c +++ b/source/dnode/vnode/tq/src/tqMetaStore.c @@ -14,6 +14,7 @@ */ #include "tqMetaStore.h" //TODO:replace by an abstract file layer +#include "osDir.h" #include #include #include @@ -22,8 +23,8 @@ #define TQ_IDX_NAME "tq.idx" -static int32_t tqHandlePutCommitted(TqMetaStore*, int64_t key, void* value); -static TqMetaHandle* tqHandleGetUncommitted(TqMetaStore*, int64_t key); +static int32_t tqHandlePutCommitted(TqMetaStore*, int64_t key, void* value); +static void* tqHandleGetUncommitted(TqMetaStore*, int64_t key); typedef struct TqMetaPageBuf { int16_t offset; @@ -31,23 +32,28 @@ typedef struct TqMetaPageBuf { } TqMetaPageBuf; TqMetaStore* tqStoreOpen(const char* path, - int serializer(TqGroupHandle*, void**), - const void* deserializer(const void*, TqGroupHandle*), - void deleter(void*)) { + int serializer(const void* pObj, void** ppBytes), + const void* deserializer(const void* pBytes, void** ppObj), + void deleter(void* pObj)) { TqMetaStore* pMeta = malloc(sizeof(TqMetaStore)); if(pMeta == NULL) { //close return NULL; } + memset(pMeta, 0, sizeof(TqMetaStore)); //concat data file name and index file name size_t pathLen = strlen(path); char name[pathLen+10]; strcpy(name, path); + if(!taosDirExist(name) && !taosMkDir(name)) { + ASSERT(false); + } strcat(name, "/" TQ_IDX_NAME); - int idxFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); + int idxFd = open(name, O_RDWR | O_CREAT, 0755); if(idxFd < 0) { + ASSERT(false); //close file //free memory return NULL; @@ -56,17 +62,24 @@ TqMetaStore* tqStoreOpen(const char* path, pMeta->idxFd = idxFd; pMeta->unpersistHead = malloc(sizeof(TqMetaList)); if(pMeta->unpersistHead == NULL) { + ASSERT(false); //close file //free memory return NULL; } + memset(pMeta->unpersistHead, 0, sizeof(TqMetaList)); + pMeta->unpersistHead->unpersistNext + = pMeta->unpersistHead->unpersistPrev + = pMeta->unpersistHead; strcpy(name, path); strcat(name, "/" TQ_META_NAME); - int fileFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); - if(fileFd < 0) return NULL; + int fileFd = open(name, O_RDWR | O_CREAT, 0755); + if(fileFd < 0){ + ASSERT(false); + return NULL; + } - memset(pMeta, 0, sizeof(TqMetaStore)); pMeta->fileFd = fileFd; pMeta->serializer = serializer; @@ -74,23 +87,47 @@ TqMetaStore* tqStoreOpen(const char* path, pMeta->deleter = deleter; //read idx file and load into memory - char readBuf[TQ_PAGE_SIZE]; - int readSize; - while((readSize = read(idxFd, readBuf, TQ_PAGE_SIZE)) != -1) { + char idxBuf[TQ_PAGE_SIZE]; + char* dataBuf = malloc(TQ_PAGE_SIZE); + if(dataBuf == NULL) { + //TODO:memory insufficient + } + int dataBufSize = TQ_PAGE_SIZE; + int idxRead, dataReadSize; + while((idxRead = read(idxFd, idxBuf, TQ_PAGE_SIZE))) { + if(idxRead == -1) { + //TODO: handle error + ASSERT(false); + } //loop read every entry - for(int i = 0; i < readSize; i += TQ_IDX_ENTRY_SIZE) { - TqMetaList *pNode = malloc(sizeof(TqMetaHandle)); - memset(pNode, 0, sizeof(TqMetaList)); + for(int i = 0; i < idxRead; i += TQ_IDX_ENTRY_SIZE) { + TqMetaList *pNode = malloc(sizeof(TqMetaList)); if(pNode == NULL) { //TODO: free memory and return error } - memcpy(&pNode->handle, &readBuf[i], TQ_IDX_ENTRY_SIZE); + memset(pNode, 0, sizeof(TqMetaList)); + memcpy(&pNode->handle, &idxBuf[i], TQ_IDX_ENTRY_SIZE); + lseek(fileFd, pNode->handle.offset, SEEK_CUR); + if(dataBufSize < pNode->handle.serializedSize) { + void *ptr = realloc(dataBuf, pNode->handle.serializedSize); + if(ptr == NULL) { + //TODO: memory insufficient + } + dataBuf = ptr; + dataBufSize = pNode->handle.serializedSize; + } + if(read(fileFd, dataBuf, pNode->handle.serializedSize) != pNode->handle.serializedSize) { + //TODO: read error + } + pMeta->deserializer(dataBuf, &pNode->handle.valueInUse); + + //put into list int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE; pNode->next = pMeta->bucket[bucketKey]; pMeta->bucket[bucketKey] = pNode; } } - + free(dataBuf); return pMeta; } @@ -102,22 +139,23 @@ int32_t tqStoreClose(TqMetaStore* pMeta) { close(pMeta->idxFd); //free memory for(int i = 0; i < TQ_BUCKET_SIZE; i++) { - TqMetaList* node = pMeta->bucket[i]; + TqMetaList* pNode = pMeta->bucket[i]; pMeta->bucket[i] = NULL; - while(node) { - ASSERT(node->unpersistNext == NULL); - ASSERT(node->unpersistPrev == NULL); - if(node->handle.valueInTxn) { - pMeta->deleter(node->handle.valueInTxn); + while(pNode) { + ASSERT(pNode->unpersistNext == NULL); + ASSERT(pNode->unpersistPrev == NULL); + if(pNode->handle.valueInTxn) { + pMeta->deleter(pNode->handle.valueInTxn); } - if(node->handle.valueInUse) { - pMeta->deleter(node->handle.valueInUse); + if(pNode->handle.valueInUse) { + pMeta->deleter(pNode->handle.valueInUse); } - TqMetaList* next = node->next; - free(node); - node = next; + TqMetaList* next = pNode->next; + free(pNode); + pNode = next; } } + free(pMeta->unpersistHead); free(pMeta); return 0; } @@ -158,36 +196,40 @@ int32_t tqStorePersist(TqMetaStore* pMeta) { pBucketNode->next = pNode->next; if(pNode->handle.valueInUse) { pMeta->deleter(pNode->handle.valueInUse); + pNode->handle.valueInUse = NULL; } free(pNode); } } - } - //serialize - void* pBytes = NULL; - int sz = pMeta->serializer(pNode->handle.valueInUse, &pBytes); - ASSERT(pBytes != NULL); - //get current offset - //append data - int64_t offset = lseek(pMeta->fileFd, 0, SEEK_CUR); - int nBytes = write(pMeta->fileFd, pBytes, sz); - //TODO: handle error in tfile - ASSERT(nBytes == sz); - - pNode->handle.offset = offset; - pNode->handle.serializedSize = sz; - - //write idx - //TODO: endian check and convert - *(bufPtr++) = pNode->handle.key; - *(bufPtr++) = pNode->handle.offset; - *(bufPtr++) = (int64_t)sz; - if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) { - nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf)); + } else { + //TODO: do not allocate each time + //serialize + void* pBytes = NULL; + int sz = pMeta->serializer(pNode->handle.valueInUse, &pBytes); + ASSERT(pBytes != NULL); + //get current offset + //append data + int64_t offset = lseek(pMeta->fileFd, 0, SEEK_CUR); + int nBytes = write(pMeta->fileFd, pBytes, sz); + free(pBytes); //TODO: handle error in tfile - ASSERT(nBytes == sizeof(writeBuf)); - memset(writeBuf, 0, TQ_PAGE_SIZE); - bufPtr = (int64_t*)writeBuf; + ASSERT(nBytes == sz); + + pNode->handle.offset = offset; + pNode->handle.serializedSize = sz; + + //write idx + //TODO: endian check and convert + *(bufPtr++) = pNode->handle.key; + *(bufPtr++) = pNode->handle.offset; + *(bufPtr++) = (int64_t)sz; + if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) { + nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf)); + //TODO: handle error in tfile + ASSERT(nBytes == sizeof(writeBuf)); + memset(writeBuf, 0, TQ_PAGE_SIZE); + bufPtr = (int64_t*)writeBuf; + } } //remove from unpersist list @@ -216,7 +258,9 @@ static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value while(pNode) { if(pNode->handle.key == key) { //TODO: think about thread safety - pMeta->deleter(pNode->handle.valueInUse); + if(pNode->handle.valueInUse) { + pMeta->deleter(pNode->handle.valueInUse); + } //change pointer ownership pNode->handle.valueInUse = value; return 0; @@ -240,13 +284,13 @@ static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value return 0; } -TqMetaHandle* tqHandleGet(TqMetaStore* pMeta, int64_t key) { +void* tqHandleGet(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { if(pNode->handle.valueInUse != NULL) { - return &pNode->handle; + return pNode->handle.valueInUse; } else { return NULL; } @@ -263,7 +307,9 @@ int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { while(pNode) { if(pNode->handle.key == key) { //TODO: think about thread safety - pMeta->deleter(pNode->handle.valueInTxn); + if(pNode->handle.valueInTxn) { + pMeta->deleter(pNode->handle.valueInTxn); + } //change pointer ownership pNode->handle.valueInTxn = value; return 0; @@ -279,16 +325,18 @@ int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { memset(pNewNode, 0, sizeof(TqMetaList)); pNewNode->handle.key = key; pNewNode->handle.valueInTxn = value; + pNewNode->next = pMeta->bucket[bucketKey]; + pMeta->bucket[bucketKey] = pNewNode; return 0; } -static TqMetaHandle* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) { +static void* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { if(pNode->handle.valueInTxn != NULL) { - return &pNode->handle; + return pNode->handle.valueInTxn; } else { return NULL; } @@ -304,10 +352,11 @@ int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) { TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { - if(pNode->handle.valueInUse != NULL) { + if(pNode->handle.valueInUse) { pMeta->deleter(pNode->handle.valueInUse); } pNode->handle.valueInUse = pNode->handle.valueInTxn; + pNode->handle.valueInTxn = NULL; if(pNode->unpersistNext == NULL) { pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; pNode->unpersistPrev = pMeta->unpersistHead; diff --git a/source/dnode/vnode/tq/test/CMakeLists.txt b/source/dnode/vnode/tq/test/CMakeLists.txt new file mode 100644 index 0000000000..6b468497d5 --- /dev/null +++ b/source/dnode/vnode/tq/test/CMakeLists.txt @@ -0,0 +1,20 @@ +add_executable(tqTest "") +target_sources(tqTest + PRIVATE + "tqMetaTest.cpp" +) +target_include_directories(tqTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/server/vnode/tq" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) + +target_link_libraries(tqTest + tq + gtest_main +) +enable_testing() +add_test( + NAME tq_test + COMMAND tqTest +) diff --git a/source/dnode/vnode/tq/test/tqMetaTest.cpp b/source/dnode/vnode/tq/test/tqMetaTest.cpp new file mode 100644 index 0000000000..e993003a78 --- /dev/null +++ b/source/dnode/vnode/tq/test/tqMetaTest.cpp @@ -0,0 +1,97 @@ +#include +#include +#include +#include + +#include "tqMetaStore.h" + +struct Foo { + int32_t a; +}; + +int FooSerializer(const void* pObj, void** ppBytes) { + Foo* foo = (Foo*) pObj; + *ppBytes = realloc(*ppBytes, sizeof(int32_t)); + **(int32_t**)ppBytes = foo->a; + return sizeof(int32_t); +} + +const void* FooDeserializer(const void* pBytes, void** ppObj) { + if(*ppObj == NULL) { + *ppObj = realloc(*ppObj, sizeof(int32_t)); + } + Foo* pFoo = *(Foo**)ppObj; + pFoo->a = *(int32_t*)pBytes; + return NULL; +} + +void FooDeleter(void* pObj) { + free(pObj); +} + +class TqMetaTest : public ::testing::Test { + protected: + + void SetUp() override { + taosRemoveDir(pathName); + pMeta = tqStoreOpen(pathName, + FooSerializer, FooDeserializer, FooDeleter); + ASSERT(pMeta); + } + + void TearDown() override { + tqStoreClose(pMeta); + } + + TqMetaStore* pMeta; + const char* pathName = "/tmp/tq_test"; +}; + +TEST_F(TqMetaTest, persistTest) { + Foo* pFoo = (Foo*)malloc(sizeof(Foo)); + pFoo->a = 2; + tqHandlePut(pMeta, 1, pFoo); + Foo* pBar = (Foo*)tqHandleGet(pMeta, 1); + EXPECT_EQ(pBar == NULL, true); + tqHandleCommit(pMeta, 1); + pBar = (Foo*)tqHandleGet(pMeta, 1); + EXPECT_EQ(pBar->a, pFoo->a); + pBar = (Foo*)tqHandleGet(pMeta, 2); + EXPECT_EQ(pBar == NULL, true); + + tqStoreClose(pMeta); + pMeta = tqStoreOpen(pathName, + FooSerializer, FooDeserializer, FooDeleter); + ASSERT(pMeta); + + pBar = (Foo*)tqHandleGet(pMeta, 1); + ASSERT_EQ(pBar != NULL, true); + EXPECT_EQ(pBar->a, 2); + + pBar = (Foo*)tqHandleGet(pMeta, 2); + EXPECT_EQ(pBar == NULL, true); + + //taosRemoveDir(pathName); +} + +TEST_F(TqMetaTest, uncommittedTest) { + Foo* pFoo = (Foo*)malloc(sizeof(Foo)); + pFoo->a = 3; + tqHandlePut(pMeta, 1, pFoo); + + pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); +} + +TEST_F(TqMetaTest, abortTest) { + Foo* pFoo = (Foo*)malloc(sizeof(Foo)); + pFoo->a = 3; + tqHandlePut(pMeta, 1, pFoo); + + pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); + + tqHandleAbort(pMeta, 1); + pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); +} diff --git a/source/dnode/vnode/tq/test/tqTests.cpp b/source/dnode/vnode/tq/test/tqTests.cpp deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/source/os/src/osDir.c b/source/os/src/osDir.c index 17ab88edf6..3b3e04378f 100644 --- a/source/os/src/osDir.c +++ b/source/os/src/osDir.c @@ -55,7 +55,7 @@ void taosRemoveDir(const char *dirname) { closedir(dir); rmdir(dirname); - printf("dir:%s is removed", dirname); + printf("dir:%s is removed\n", dirname); } bool taosDirExist(char *dirname) { return access(dirname, F_OK) == 0; } @@ -138,4 +138,4 @@ bool taosRealPath(char *dirname, int32_t maxlen) { return true; } -#endif \ No newline at end of file +#endif From f5992851891ea30a951e0c68e04bcb048bd28158 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Thu, 4 Nov 2021 10:16:59 +0800 Subject: [PATCH 2/3] add tqHandleCopyPut interface --- source/dnode/vnode/tq/inc/tqMetaStore.h | 4 +- source/dnode/vnode/tq/src/tqMetaStore.c | 67 ++++++++++++++++++++++- source/dnode/vnode/tq/test/tqMetaTest.cpp | 15 ++++- 3 files changed, 79 insertions(+), 7 deletions(-) diff --git a/source/dnode/vnode/tq/inc/tqMetaStore.h b/source/dnode/vnode/tq/inc/tqMetaStore.h index f3918d0a8e..f8700e065f 100644 --- a/source/dnode/vnode/tq/inc/tqMetaStore.h +++ b/source/dnode/vnode/tq/inc/tqMetaStore.h @@ -60,6 +60,7 @@ typedef struct TqMetaStore { TqMetaList* unpersistHead; int fileFd; //TODO:temporaral use, to be replaced by unified tfile int idxFd; //TODO:temporaral use, to be replaced by unified tfile + char* dirPath; int (*serializer)(const void* pObj, void** ppBytes); const void* (*deserializer)(const void* pBytes, void** ppObj); void (*deleter)(void*); @@ -75,7 +76,8 @@ int32_t tqStoreClose(TqMetaStore*); int32_t tqStorePersist(TqMetaStore*); void* tqHandleGet(TqMetaStore*, int64_t key); -int32_t tqHandlePut(TqMetaStore*, int64_t key, void* value); +int32_t tqHandleMovePut(TqMetaStore*, int64_t key, void* value); +int32_t tqHandleCopyPut(TqMetaStore*, int64_t key, void* value, size_t vsize); //do commit int32_t tqHandleCommit(TqMetaStore*, int64_t key); //delete uncommitted diff --git a/source/dnode/vnode/tq/src/tqMetaStore.c b/source/dnode/vnode/tq/src/tqMetaStore.c index f850f73e12..5dac02ff67 100644 --- a/source/dnode/vnode/tq/src/tqMetaStore.c +++ b/source/dnode/vnode/tq/src/tqMetaStore.c @@ -44,6 +44,12 @@ TqMetaStore* tqStoreOpen(const char* path, //concat data file name and index file name size_t pathLen = strlen(path); + pMeta->dirPath = malloc(pathLen+1); + if(pMeta->dirPath != NULL) { + //TODO: memory insufficient + } + strcpy(pMeta->dirPath, path); + char name[pathLen+10]; strcpy(name, path); @@ -155,15 +161,35 @@ int32_t tqStoreClose(TqMetaStore* pMeta) { pNode = next; } } + free(pMeta->dirPath); free(pMeta->unpersistHead); free(pMeta); return 0; } int32_t tqStoreDelete(TqMetaStore* pMeta) { - //close file - //delete file + close(pMeta->fileFd); + close(pMeta->idxFd); //free memory + for(int i = 0; i < TQ_BUCKET_SIZE; i++) { + TqMetaList* pNode = pMeta->bucket[i]; + pMeta->bucket[i] = NULL; + while(pNode) { + if(pNode->handle.valueInTxn) { + pMeta->deleter(pNode->handle.valueInTxn); + } + if(pNode->handle.valueInUse) { + pMeta->deleter(pNode->handle.valueInUse); + } + TqMetaList* next = pNode->next; + free(pNode); + pNode = next; + } + } + free(pMeta->unpersistHead); + taosRemoveDir(pMeta->dirPath); + free(pMeta->dirPath); + free(pMeta); return 0; } @@ -301,7 +327,7 @@ void* tqHandleGet(TqMetaStore* pMeta, int64_t key) { return NULL; } -int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { +int32_t tqHandleMovePut(TqMetaStore* pMeta, int64_t key, void* value) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { @@ -330,6 +356,41 @@ int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { return 0; } +int32_t tqHandleCopyPut(TqMetaStore* pMeta, int64_t key, void* value, size_t vsize) { + void *vmem = malloc(vsize); + if(vmem == NULL) { + //TODO: memory error + return -1; + } + memcpy(vmem, value, vsize); + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + //TODO: think about thread safety + if(pNode->handle.valueInTxn) { + pMeta->deleter(pNode->handle.valueInTxn); + } + //change pointer ownership + pNode->handle.valueInTxn = vmem; + return 0; + } else { + pNode = pNode->next; + } + } + TqMetaList *pNewNode = malloc(sizeof(TqMetaList)); + if(pNewNode == NULL) { + //TODO: memory error + return -1; + } + memset(pNewNode, 0, sizeof(TqMetaList)); + pNewNode->handle.key = key; + pNewNode->handle.valueInTxn = vmem; + pNewNode->next = pMeta->bucket[bucketKey]; + pMeta->bucket[bucketKey] = pNewNode; + return 0; +} + static void* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; diff --git a/source/dnode/vnode/tq/test/tqMetaTest.cpp b/source/dnode/vnode/tq/test/tqMetaTest.cpp index e993003a78..8afb10e824 100644 --- a/source/dnode/vnode/tq/test/tqMetaTest.cpp +++ b/source/dnode/vnode/tq/test/tqMetaTest.cpp @@ -47,10 +47,19 @@ class TqMetaTest : public ::testing::Test { const char* pathName = "/tmp/tq_test"; }; +TEST_F(TqMetaTest, copyPutTest) { + Foo foo; + foo.a = 3; + tqHandleCopyPut(pMeta, 1, &foo, sizeof(Foo)); + + Foo* pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); +} + TEST_F(TqMetaTest, persistTest) { Foo* pFoo = (Foo*)malloc(sizeof(Foo)); pFoo->a = 2; - tqHandlePut(pMeta, 1, pFoo); + tqHandleMovePut(pMeta, 1, pFoo); Foo* pBar = (Foo*)tqHandleGet(pMeta, 1); EXPECT_EQ(pBar == NULL, true); tqHandleCommit(pMeta, 1); @@ -77,7 +86,7 @@ TEST_F(TqMetaTest, persistTest) { TEST_F(TqMetaTest, uncommittedTest) { Foo* pFoo = (Foo*)malloc(sizeof(Foo)); pFoo->a = 3; - tqHandlePut(pMeta, 1, pFoo); + tqHandleMovePut(pMeta, 1, pFoo); pFoo = (Foo*) tqHandleGet(pMeta, 1); EXPECT_EQ(pFoo == NULL, true); @@ -86,7 +95,7 @@ TEST_F(TqMetaTest, uncommittedTest) { TEST_F(TqMetaTest, abortTest) { Foo* pFoo = (Foo*)malloc(sizeof(Foo)); pFoo->a = 3; - tqHandlePut(pMeta, 1, pFoo); + tqHandleMovePut(pMeta, 1, pFoo); pFoo = (Foo*) tqHandleGet(pMeta, 1); EXPECT_EQ(pFoo == NULL, true); From 7c2048e7df47e16ac3430a6ff90d1dae2815aafd Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 5 Nov 2021 15:42:28 +0800 Subject: [PATCH 3/3] implement tqOpen and tqPersist --- include/server/vnode/tq/tq.h | 3 - source/dnode/vnode/tq/inc/tqMetaStore.h | 29 ++- source/dnode/vnode/tq/src/tqMetaStore.c | 287 ++++++++++++++-------- source/dnode/vnode/tq/test/tqMetaTest.cpp | 39 ++- 4 files changed, 247 insertions(+), 111 deletions(-) diff --git a/include/server/vnode/tq/tq.h b/include/server/vnode/tq/tq.h index 4a93491056..3aeaf9acb6 100644 --- a/include/server/vnode/tq/tq.h +++ b/include/server/vnode/tq/tq.h @@ -19,9 +19,6 @@ #include "os.h" #include "tutil.h" -#define TQ_ACTION_INSERT 0x7f7f7f7fULL -#define TQ_ACTION_DELETE 0x80808080ULL - #ifdef __cplusplus extern "C" { #endif diff --git a/source/dnode/vnode/tq/inc/tqMetaStore.h b/source/dnode/vnode/tq/inc/tqMetaStore.h index f8700e065f..52cc767409 100644 --- a/source/dnode/vnode/tq/inc/tqMetaStore.h +++ b/source/dnode/vnode/tq/inc/tqMetaStore.h @@ -37,6 +37,24 @@ inline static int TqEmptyTail() { //16 return TQ_PAGE_SIZE - TqMaxEntryOnePage(); } +#define TQ_ACTION_CONST 0 +#define TQ_ACTION_INUSE 1 +#define TQ_ACTION_INUSE_CONT 2 +#define TQ_ACTION_INTXN 3 + +#define TQ_SVER 0 + +static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST; +#define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE + +typedef struct TqSerializedHead { + int16_t ver; + int16_t action; + int32_t checksum; + int64_t ssize; + char content[]; +} TqSerializedHead; + typedef struct TqMetaHandle { int64_t key; int64_t offset; @@ -61,14 +79,14 @@ typedef struct TqMetaStore { int fileFd; //TODO:temporaral use, to be replaced by unified tfile int idxFd; //TODO:temporaral use, to be replaced by unified tfile char* dirPath; - int (*serializer)(const void* pObj, void** ppBytes); - const void* (*deserializer)(const void* pBytes, void** ppObj); + int (*serializer)(const void* pObj, TqSerializedHead** ppHead); + const void* (*deserializer)(const TqSerializedHead* pHead, void** ppObj); void (*deleter)(void*); } TqMetaStore; TqMetaStore* tqStoreOpen(const char* path, - int serializer(const void* pObj, void** ppBytes), - const void* deserializer(const void* pBytes, void** ppObj), + int serializer(const void* pObj, TqSerializedHead** ppHead), + const void* deserializer(const TqSerializedHead* pHead, void** ppObj), void deleter(void* pObj)); int32_t tqStoreClose(TqMetaStore*); //int32_t tqStoreDelete(TqMetaStore*); @@ -82,7 +100,8 @@ int32_t tqHandleCopyPut(TqMetaStore*, int64_t key, void* value, size_t vsize); int32_t tqHandleCommit(TqMetaStore*, int64_t key); //delete uncommitted int32_t tqHandleAbort(TqMetaStore*, int64_t key); -//delete committed +//delete committed kv pair +//notice that a delete action still needs to be committed int32_t tqHandleDel(TqMetaStore*, int64_t key); //delete both committed and uncommitted int32_t tqHandleClear(TqMetaStore*, int64_t key); diff --git a/source/dnode/vnode/tq/src/tqMetaStore.c b/source/dnode/vnode/tq/src/tqMetaStore.c index 5dac02ff67..a4c2b90491 100644 --- a/source/dnode/vnode/tq/src/tqMetaStore.c +++ b/source/dnode/vnode/tq/src/tqMetaStore.c @@ -26,14 +26,24 @@ static int32_t tqHandlePutCommitted(TqMetaStore*, int64_t key, void* value); static void* tqHandleGetUncommitted(TqMetaStore*, int64_t key); +static inline void tqLinkUnpersist(TqMetaStore *pMeta, TqMetaList* pNode) { + if(pNode->unpersistNext == NULL) { + pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; + pNode->unpersistPrev = pMeta->unpersistHead; + pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; + pMeta->unpersistHead->unpersistNext = pNode; + } +} + + typedef struct TqMetaPageBuf { int16_t offset; char buffer[TQ_PAGE_SIZE]; } TqMetaPageBuf; TqMetaStore* tqStoreOpen(const char* path, - int serializer(const void* pObj, void** ppBytes), - const void* deserializer(const void* pBytes, void** ppObj), + int serializer(const void* pObj, TqSerializedHead** ppHead), + const void* deserializer(const TqSerializedHead* pHead, void** ppObj), void deleter(void* pObj)) { TqMetaStore* pMeta = malloc(sizeof(TqMetaStore)); if(pMeta == NULL) { @@ -94,12 +104,12 @@ TqMetaStore* tqStoreOpen(const char* path, //read idx file and load into memory char idxBuf[TQ_PAGE_SIZE]; - char* dataBuf = malloc(TQ_PAGE_SIZE); - if(dataBuf == NULL) { + TqSerializedHead* serializedObj = malloc(TQ_PAGE_SIZE); + if(serializedObj == NULL) { //TODO:memory insufficient } - int dataBufSize = TQ_PAGE_SIZE; - int idxRead, dataReadSize; + int idxRead; + int allocated = TQ_PAGE_SIZE; while((idxRead = read(idxFd, idxBuf, TQ_PAGE_SIZE))) { if(idxRead == -1) { //TODO: handle error @@ -114,26 +124,82 @@ TqMetaStore* tqStoreOpen(const char* path, memset(pNode, 0, sizeof(TqMetaList)); memcpy(&pNode->handle, &idxBuf[i], TQ_IDX_ENTRY_SIZE); lseek(fileFd, pNode->handle.offset, SEEK_CUR); - if(dataBufSize < pNode->handle.serializedSize) { - void *ptr = realloc(dataBuf, pNode->handle.serializedSize); + if(allocated < pNode->handle.serializedSize) { + void *ptr = realloc(serializedObj, pNode->handle.serializedSize); if(ptr == NULL) { //TODO: memory insufficient } - dataBuf = ptr; - dataBufSize = pNode->handle.serializedSize; + serializedObj = ptr; + allocated = pNode->handle.serializedSize; } - if(read(fileFd, dataBuf, pNode->handle.serializedSize) != pNode->handle.serializedSize) { + serializedObj->ssize = pNode->handle.serializedSize; + if(read(fileFd, serializedObj, pNode->handle.serializedSize) != pNode->handle.serializedSize) { //TODO: read error } - pMeta->deserializer(dataBuf, &pNode->handle.valueInUse); + if(serializedObj->action == TQ_ACTION_INUSE) { + if(serializedObj->ssize != sizeof(TqSerializedHead)) { + pMeta->deserializer(serializedObj, &pNode->handle.valueInUse); + } else { + pNode->handle.valueInUse = TQ_DELETE_TOKEN; + } + } else if(serializedObj->action == TQ_ACTION_INTXN) { + if(serializedObj->ssize != sizeof(TqSerializedHead)) { + pMeta->deserializer(serializedObj, &pNode->handle.valueInTxn); + } else { + pNode->handle.valueInTxn = TQ_DELETE_TOKEN; + } + } else if(serializedObj->action == TQ_ACTION_INUSE_CONT) { + if(serializedObj->ssize != sizeof(TqSerializedHead)) { + pMeta->deserializer(serializedObj, &pNode->handle.valueInUse); + } else { + pNode->handle.valueInUse = TQ_DELETE_TOKEN; + } + serializedObj = POINTER_SHIFT(serializedObj, serializedObj->ssize); + if(serializedObj->ssize != sizeof(TqSerializedHead)) { + pMeta->deserializer(serializedObj, &pNode->handle.valueInTxn); + } else { + pNode->handle.valueInTxn = TQ_DELETE_TOKEN; + } + } else { + ASSERT(0); + } //put into list int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE; - pNode->next = pMeta->bucket[bucketKey]; - pMeta->bucket[bucketKey] = pNode; + TqMetaList* pBucketNode = pMeta->bucket[bucketKey]; + if(pBucketNode == NULL) { + pMeta->bucket[bucketKey] = pNode; + } else if(pBucketNode->handle.key == pNode->handle.key) { + pNode->next = pBucketNode->next; + pMeta->bucket[bucketKey] = pNode; + } else { + while(pBucketNode->next && + pBucketNode->next->handle.key == pNode->handle.key) { + pBucketNode = pBucketNode->next; + } + if(pBucketNode->next) { + ASSERT(pBucketNode->next->handle.key == pNode->handle.key); + TqMetaList *pNodeTmp = pBucketNode->next; + pBucketNode->next = pNodeTmp->next; + pBucketNode = pNodeTmp; + } else { + pBucketNode = NULL; + } + } + if(pBucketNode) { + if(pBucketNode->handle.valueInUse + && pBucketNode->handle.valueInUse != TQ_DELETE_TOKEN) { + pMeta->deleter(pBucketNode->handle.valueInUse); + } + if(pBucketNode->handle.valueInTxn + && pBucketNode->handle.valueInTxn != TQ_DELETE_TOKEN) { + pMeta->deleter(pBucketNode->handle.valueInTxn); + } + free(pBucketNode); + } } } - free(dataBuf); + free(serializedObj); return pMeta; } @@ -146,14 +212,15 @@ int32_t tqStoreClose(TqMetaStore* pMeta) { //free memory for(int i = 0; i < TQ_BUCKET_SIZE; i++) { TqMetaList* pNode = pMeta->bucket[i]; - pMeta->bucket[i] = NULL; while(pNode) { ASSERT(pNode->unpersistNext == NULL); ASSERT(pNode->unpersistPrev == NULL); - if(pNode->handle.valueInTxn) { + if(pNode->handle.valueInTxn + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInTxn); } - if(pNode->handle.valueInUse) { + if(pNode->handle.valueInUse + && pNode->handle.valueInUse != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInUse); } TqMetaList* next = pNode->next; @@ -175,10 +242,12 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) { TqMetaList* pNode = pMeta->bucket[i]; pMeta->bucket[i] = NULL; while(pNode) { - if(pNode->handle.valueInTxn) { + if(pNode->handle.valueInTxn + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInTxn); } - if(pNode->handle.valueInUse) { + if(pNode->handle.valueInUse + && pNode->handle.valueInUse != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInUse); } TqMetaList* next = pNode->next; @@ -199,73 +268,89 @@ int32_t tqStorePersist(TqMetaStore* pMeta) { int64_t* bufPtr = (int64_t*)writeBuf; TqMetaList *pHead = pMeta->unpersistHead; TqMetaList *pNode = pHead->unpersistNext; + TqSerializedHead *pSHead = malloc(sizeof(TqSerializedHead)); + if(pSHead == NULL) { + //TODO: memory error + return -1; + } + pSHead->ver = TQ_SVER; + pSHead->checksum = 0; + pSHead->ssize = sizeof(TqSerializedHead); + int allocatedSize = sizeof(TqSerializedHead); + int offset = lseek(pMeta->fileFd, 0, SEEK_CUR); while(pHead != pNode) { - if(pNode->handle.valueInUse == NULL) { - //put delete token in data file - uint32_t delete = TQ_ACTION_DELETE; - int nBytes = write(pMeta->fileFd, &delete, sizeof(uint32_t)); - ASSERT(nBytes == sizeof(uint32_t)); + int nBytes = 0; - //remove from list + if(pNode->handle.valueInUse) { + if(pNode->handle.valueInTxn) { + pSHead->action = TQ_ACTION_INUSE_CONT; + } else { + pSHead->action = TQ_ACTION_INUSE; + } + + if(pNode->handle.valueInUse == TQ_DELETE_TOKEN) { + pSHead->ssize = sizeof(TqSerializedHead); + } else { + pMeta->serializer(pNode->handle.valueInUse, &pSHead); + } + nBytes = write(pMeta->fileFd, pSHead, pSHead->ssize); + ASSERT(nBytes == pSHead->ssize); + } + + if(pNode->handle.valueInTxn) { + pSHead->action = TQ_ACTION_INTXN; + if(pNode->handle.valueInTxn == TQ_DELETE_TOKEN) { + pSHead->ssize = sizeof(TqSerializedHead); + } else { + pMeta->serializer(pNode->handle.valueInTxn, &pSHead); + } + int nBytesTxn = write(pMeta->fileFd, pSHead, pSHead->ssize); + ASSERT(nBytesTxn == pSHead->ssize); + nBytes += nBytesTxn; + } + + //write idx file + //TODO: endian check and convert + *(bufPtr++) = pNode->handle.key; + *(bufPtr++) = pNode->handle.offset; + *(bufPtr++) = (int64_t)nBytes; + if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) { + nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf)); + //TODO: handle error with tfile + ASSERT(nBytes == sizeof(writeBuf)); + memset(writeBuf, 0, TQ_PAGE_SIZE); + bufPtr = (int64_t*)writeBuf; + } + //remove from unpersist list + pHead->unpersistNext = pNode->unpersistNext; + pHead->unpersistNext->unpersistPrev = pHead; + pNode->unpersistPrev = pNode->unpersistNext = NULL; + pNode = pHead->unpersistNext; + + //remove from bucket + if(pNode->handle.valueInUse == TQ_DELETE_TOKEN && + pNode->handle.valueInTxn == NULL + ) { int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE; TqMetaList* pBucketHead = pMeta->bucket[bucketKey]; if(pBucketHead == pNode) { - pMeta->bucket[bucketKey] = pBucketHead->next; + pMeta->bucket[bucketKey] = pNode->next; } else { TqMetaList* pBucketNode = pBucketHead; while(pBucketNode->next != NULL && pBucketNode->next != pNode) { pBucketNode = pBucketNode->next; } - if(pBucketNode->next != NULL) { - ASSERT(pBucketNode->next == pNode); - pBucketNode->next = pNode->next; - if(pNode->handle.valueInUse) { - pMeta->deleter(pNode->handle.valueInUse); - pNode->handle.valueInUse = NULL; - } - free(pNode); - } - } - } else { - //TODO: do not allocate each time - //serialize - void* pBytes = NULL; - int sz = pMeta->serializer(pNode->handle.valueInUse, &pBytes); - ASSERT(pBytes != NULL); - //get current offset - //append data - int64_t offset = lseek(pMeta->fileFd, 0, SEEK_CUR); - int nBytes = write(pMeta->fileFd, pBytes, sz); - free(pBytes); - //TODO: handle error in tfile - ASSERT(nBytes == sz); - - pNode->handle.offset = offset; - pNode->handle.serializedSize = sz; - - //write idx - //TODO: endian check and convert - *(bufPtr++) = pNode->handle.key; - *(bufPtr++) = pNode->handle.offset; - *(bufPtr++) = (int64_t)sz; - if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) { - nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf)); - //TODO: handle error in tfile - ASSERT(nBytes == sizeof(writeBuf)); - memset(writeBuf, 0, TQ_PAGE_SIZE); - bufPtr = (int64_t*)writeBuf; + //impossible for pBucket->next == NULL + ASSERT(pBucketNode->next == pNode); + pBucketNode->next = pNode->next; } + free(pNode); } - - //remove from unpersist list - pHead->unpersistNext = pNode->unpersistNext; - pHead->unpersistNext->unpersistPrev = pHead; - - pNode->unpersistPrev = pNode->unpersistNext = NULL; - pNode = pHead->unpersistNext; } + //write left bytes + free(pSHead); if((char*)bufPtr != writeBuf) { int used = (char*)bufPtr - writeBuf; int nBytes = write(pMeta->idxFd, writeBuf, used); @@ -284,7 +369,8 @@ static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value while(pNode) { if(pNode->handle.key == key) { //TODO: think about thread safety - if(pNode->handle.valueInUse) { + if(pNode->handle.valueInUse + && pNode->handle.valueInUse != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInUse); } //change pointer ownership @@ -333,11 +419,13 @@ int32_t tqHandleMovePut(TqMetaStore* pMeta, int64_t key, void* value) { while(pNode) { if(pNode->handle.key == key) { //TODO: think about thread safety - if(pNode->handle.valueInTxn) { + if(pNode->handle.valueInTxn + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInTxn); } //change pointer ownership pNode->handle.valueInTxn = value; + tqLinkUnpersist(pMeta, pNode); return 0; } else { pNode = pNode->next; @@ -353,6 +441,7 @@ int32_t tqHandleMovePut(TqMetaStore* pMeta, int64_t key, void* value) { pNewNode->handle.valueInTxn = value; pNewNode->next = pMeta->bucket[bucketKey]; pMeta->bucket[bucketKey] = pNewNode; + tqLinkUnpersist(pMeta, pNewNode); return 0; } @@ -368,11 +457,13 @@ int32_t tqHandleCopyPut(TqMetaStore* pMeta, int64_t key, void* value, size_t vsi while(pNode) { if(pNode->handle.key == key) { //TODO: think about thread safety - if(pNode->handle.valueInTxn) { + if(pNode->handle.valueInTxn + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInTxn); } //change pointer ownership pNode->handle.valueInTxn = vmem; + tqLinkUnpersist(pMeta, pNode); return 0; } else { pNode = pNode->next; @@ -388,6 +479,7 @@ int32_t tqHandleCopyPut(TqMetaStore* pMeta, int64_t key, void* value, size_t vsi pNewNode->handle.valueInTxn = vmem; pNewNode->next = pMeta->bucket[bucketKey]; pMeta->bucket[bucketKey] = pNewNode; + tqLinkUnpersist(pMeta, pNewNode); return 0; } @@ -396,7 +488,8 @@ static void* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) { TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { - if(pNode->handle.valueInTxn != NULL) { + if(pNode->handle.valueInTxn != NULL + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { return pNode->handle.valueInTxn; } else { return NULL; @@ -413,17 +506,13 @@ int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) { TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { - if(pNode->handle.valueInUse) { + if(pNode->handle.valueInUse + && pNode->handle.valueInUse != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInUse); } pNode->handle.valueInUse = pNode->handle.valueInTxn; pNode->handle.valueInTxn = NULL; - if(pNode->unpersistNext == NULL) { - pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; - pNode->unpersistPrev = pMeta->unpersistHead; - pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; - pMeta->unpersistHead->unpersistNext = pNode; - } + tqLinkUnpersist(pMeta, pNode); return 0; } else { pNode = pNode->next; @@ -437,9 +526,12 @@ int32_t tqHandleAbort(TqMetaStore* pMeta, int64_t key) { TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { - if(pNode->handle.valueInTxn != NULL) { - pMeta->deleter(pNode->handle.valueInTxn); + if(pNode->handle.valueInTxn) { + if(pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { + pMeta->deleter(pNode->handle.valueInTxn); + } pNode->handle.valueInTxn = NULL; + tqLinkUnpersist(pMeta, pNode); return 0; } return -1; @@ -454,9 +546,11 @@ int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { - if(pNode->handle.key == key) { + if(pNode->handle.valueInTxn + && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { pMeta->deleter(pNode->handle.valueInTxn); - pNode->handle.valueInTxn = NULL; + pNode->handle.valueInTxn = TQ_DELETE_TOKEN; + tqLinkUnpersist(pMeta, pNode); return 0; } else { pNode = pNode->next; @@ -474,21 +568,20 @@ int32_t tqHandleClear(TqMetaStore* pMeta, int64_t key) { if(pNode->handle.key == key) { if(pNode->handle.valueInUse != NULL) { exist = true; - pMeta->deleter(pNode->handle.valueInUse); - pNode->handle.valueInUse = NULL; + if(pNode->handle.valueInUse != TQ_DELETE_TOKEN) { + pMeta->deleter(pNode->handle.valueInUse); + } + pNode->handle.valueInUse = TQ_DELETE_TOKEN; } if(pNode->handle.valueInTxn != NULL) { exist = true; - pMeta->deleter(pNode->handle.valueInTxn); - pNode->handle.valueInTxn = NULL; + if(pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { + pMeta->deleter(pNode->handle.valueInTxn); + } + pNode->handle.valueInTxn = TQ_DELETE_TOKEN; } if(exist) { - if(pNode->unpersistNext == NULL) { - pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; - pNode->unpersistPrev = pMeta->unpersistHead; - pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; - pMeta->unpersistHead->unpersistNext = pNode; - } + tqLinkUnpersist(pMeta, pNode); return 0; } return -1; diff --git a/source/dnode/vnode/tq/test/tqMetaTest.cpp b/source/dnode/vnode/tq/test/tqMetaTest.cpp index 8afb10e824..20a0368c4c 100644 --- a/source/dnode/vnode/tq/test/tqMetaTest.cpp +++ b/source/dnode/vnode/tq/test/tqMetaTest.cpp @@ -9,19 +9,22 @@ struct Foo { int32_t a; }; -int FooSerializer(const void* pObj, void** ppBytes) { +int FooSerializer(const void* pObj, TqSerializedHead** ppHead) { Foo* foo = (Foo*) pObj; - *ppBytes = realloc(*ppBytes, sizeof(int32_t)); - **(int32_t**)ppBytes = foo->a; - return sizeof(int32_t); + if((*ppHead) == NULL || (*ppHead)->ssize < sizeof(TqSerializedHead) + sizeof(int32_t)) { + *ppHead = (TqSerializedHead*)realloc(*ppHead, sizeof(TqSerializedHead) + sizeof(int32_t)); + (*ppHead)->ssize = sizeof(TqSerializedHead) + sizeof(int32_t); + } + *(int32_t*)(*ppHead)->content = foo->a; + return (*ppHead)->ssize; } -const void* FooDeserializer(const void* pBytes, void** ppObj) { +const void* FooDeserializer(const TqSerializedHead* pHead, void** ppObj) { if(*ppObj == NULL) { *ppObj = realloc(*ppObj, sizeof(int32_t)); } Foo* pFoo = *(Foo**)ppObj; - pFoo->a = *(int32_t*)pBytes; + pFoo->a = *(int32_t*)pHead->content; return NULL; } @@ -104,3 +107,27 @@ TEST_F(TqMetaTest, abortTest) { pFoo = (Foo*) tqHandleGet(pMeta, 1); EXPECT_EQ(pFoo == NULL, true); } + +TEST_F(TqMetaTest, deleteTest) { + Foo* pFoo = (Foo*)malloc(sizeof(Foo)); + pFoo->a = 3; + tqHandleMovePut(pMeta, 1, pFoo); + + pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); + + tqHandleCommit(pMeta, 1); + + pFoo = (Foo*) tqHandleGet(pMeta, 1); + ASSERT_EQ(pFoo != NULL, true); + EXPECT_EQ(pFoo->a, 3); + + tqHandleDel(pMeta, 1); + pFoo = (Foo*) tqHandleGet(pMeta, 1); + ASSERT_EQ(pFoo != NULL, true); + EXPECT_EQ(pFoo->a, 3); + + tqHandleCommit(pMeta, 1); + pFoo = (Foo*) tqHandleGet(pMeta, 1); + EXPECT_EQ(pFoo == NULL, true); +}