From 6d55ee00efc95d4c69d99467b3ecc390507c6e85 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 1 Nov 2021 16:14:55 +0800 Subject: [PATCH 1/3] implement tq meta --- source/server/vnode/tq/inc/tqMetaStore.h | 33 ++-- source/server/vnode/tq/src/tq.c | 26 +-- source/server/vnode/tq/src/tqMetaStore.c | 226 +++++++++++++++++++++-- 3 files changed, 241 insertions(+), 44 deletions(-) diff --git a/source/server/vnode/tq/inc/tqMetaStore.h b/source/server/vnode/tq/inc/tqMetaStore.h index 066e76028d..a82bf2a4a9 100644 --- a/source/server/vnode/tq/inc/tqMetaStore.h +++ b/source/server/vnode/tq/inc/tqMetaStore.h @@ -17,9 +17,9 @@ #define _TQ_META_STORE_H_ #include "os.h" +#include "tq.h" -#define TQ_INUSE_SIZE 0xFF -#define TQ_PAGE_SIZE 4096 +#define TQ_BUCKET_SIZE 0xFF #ifdef __cplusplus extern "C" { @@ -35,34 +35,39 @@ typedef struct TqMetaHandle { typedef struct TqMetaList { TqMetaHandle handle; struct TqMetaList* next; - struct TqMetaList* inTxnPrev; - struct TqMetaList* inTxnNext; + //struct TqMetaList* inTxnPrev; + //struct TqMetaList* inTxnNext; struct TqMetaList* unpersistPrev; struct TqMetaList* unpersistNext; } TqMetaList; typedef struct TqMetaStore { - TqMetaList* inUse[TQ_INUSE_SIZE]; + TqMetaList* bucket[TQ_BUCKET_SIZE]; //a table head, key is empty TqMetaList* unpersistHead; - int fileFd; //TODO:temporaral use - int idxFd; //TODO:temporaral use - void* (*serializer)(void*); - void* (*deserializer)(void*); + 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*); void (*deleter)(void*); } TqMetaStore; -TqMetaStore* tqStoreOpen(const char* path, void* serializer(void* ), void* deserializer(void*), void deleter(void*)); +TqMetaStore* tqStoreOpen(const char* path, + int serializer(TqGroupHandle*, void**), + const void* deserializer(const void*, TqGroupHandle*), + void deleter(void*)); int32_t tqStoreClose(TqMetaStore*); -int32_t tqStoreDelete(TqMetaStore*); +//int32_t tqStoreDelete(TqMetaStore*); //int32_t TqStoreCommitAll(TqMetaStore*); int32_t tqStorePersist(TqMetaStore*); TqMetaHandle* tqHandleGetInUse(TqMetaStore*, int64_t key); -int32_t tqHandlePutInUse(TqMetaStore*, TqMetaHandle* handle); +int32_t tqHandlePutInUse(TqMetaStore*, int64_t key, void* value); TqMetaHandle* tqHandleGetInTxn(TqMetaStore*, int64_t key); -int32_t tqHandlePutInTxn(TqMetaStore*, TqMetaHandle* handle); -//delete in-use-handle, make in-txn-handle in use +int32_t tqHandlePutInTxn(TqMetaStore*, int64_t key, void* value); +//will replace old handle +//int32_t tqHandlePut(TqMetaStore*, TqMetaHandle* handle); +//delete in-use-handle, and put it in use int32_t tqHandleCommit(TqMetaStore*, int64_t key); //delete in-txn-handle int32_t tqHandleAbort(TqMetaStore*, int64_t key); diff --git a/source/server/vnode/tq/src/tq.c b/source/server/vnode/tq/src/tq.c index 7ecdfe7f19..e8076515a7 100644 --- a/source/server/vnode/tq/src/tq.c +++ b/source/server/vnode/tq/src/tq.c @@ -49,12 +49,12 @@ static int tqAckOneTopic(TqBufferHandle *bhandle, TmqOneAck *pAck, TqQueryMsg** return 0; } -static int tqAck(TqGroupHandle* ghandle, TmqAcks* pAcks) { +static int tqAck(TqGroupHandle* gHandle, TmqAcks* pAcks) { int32_t ackNum = pAcks->ackNum; TmqOneAck *acks = pAcks->acks; //double ptr for acks and list int i = 0; - TqListHandle* node = ghandle->head; + TqListHandle* node = gHandle->head; int ackCnt = 0; TqQueryMsg *pQuery = NULL; while(i < ackNum && node->next) { @@ -99,8 +99,8 @@ int tqDropTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) { return 0; } -static int tqFetch(TqGroupHandle* ghandle, void** msg) { - TqListHandle* head = ghandle->head; +static int tqFetch(TqGroupHandle* gHandle, void** msg) { + TqListHandle* head = gHandle->head; TqListHandle* node = head; int totSize = 0; //TODO: make it a macro @@ -148,7 +148,7 @@ TqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) { return NULL; } -int tqLaunchQuery(TqGroupHandle* ghandle) { +int tqLaunchQuery(TqGroupHandle* gHandle) { return 0; } @@ -156,7 +156,7 @@ int tqSendLaunchQuery(TqGroupHandle* gHandle) { return 0; } -/*int tqMoveOffsetToNext(TqGroupHandle* ghandle) {*/ +/*int tqMoveOffsetToNext(TqGroupHandle* gHandle) {*/ /*return 0;*/ /*}*/ @@ -177,13 +177,13 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) { return -1; } int64_t clientId = pMsg->head.clientId; - TqGroupHandle *ghandle = tqGetGroupHandle(pTq, clientId); - if(ghandle == NULL) { + TqGroupHandle *gHandle = tqGetGroupHandle(pTq, clientId); + if(gHandle == NULL) { //client not connect return -1; } if(pMsg->acks.ackNum != 0) { - if(tqAck(ghandle, &pMsg->acks) != 0) { + if(tqAck(gHandle, &pMsg->acks) != 0) { //ack not success return -1; } @@ -191,13 +191,13 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) { TmqConsumeRsp *pRsp = (TmqConsumeRsp*) pMsg; - if(tqFetch(ghandle, (void**)&pRsp->msgs) <= 0) { + if(tqFetch(gHandle, (void**)&pRsp->msgs) <= 0) { //fetch error return -1; } //judge and launch new query - if(tqLaunchQuery(ghandle)) { + if(tqLaunchQuery(gHandle)) { //launch query error return -1; } @@ -206,7 +206,7 @@ int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) { int tqSerializeGroupHandle(TqGroupHandle *gHandle, void** ppBytes) { //calculate size - int sz = tqGetGHandleSSize(gHandle); + int sz = tqGetgHandleSSize(gHandle); void* ptr = realloc(*ppBytes, sz); if(ptr == NULL) { free(ppBytes); @@ -313,7 +313,7 @@ const void* tqDeserializeBufItem(const void* pBytes, TqBufferItem *bufItem) { } //TODO: make this a macro -int tqGetGHandleSSize(const TqGroupHandle *gHandle) { +int tqGetgHandleSSize(const TqGroupHandle *gHandle) { return sizeof(int64_t) * 2 + sizeof(int32_t) + gHandle->topicNum * tqBufHandleSSize(); diff --git a/source/server/vnode/tq/src/tqMetaStore.c b/source/server/vnode/tq/src/tqMetaStore.c index d652058c74..a93db7851b 100644 --- a/source/server/vnode/tq/src/tqMetaStore.c +++ b/source/server/vnode/tq/src/tqMetaStore.c @@ -15,17 +15,28 @@ #include "tqMetaStore.h" //TODO:replace by a abstract file layer #include +#include #include +#define TQ_PAGE_SIZE 4096 +#define TQ_META_NAME "tq.meta" +#define TQ_IDX_NAME "tq.idx" + typedef struct TqMetaPageBuf { int16_t offset; char buffer[TQ_PAGE_SIZE]; } TqMetaPageBuf; -TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*), - void* deserializer(void*), void deleter(void*)) { +TqMetaStore* tqStoreOpen(const char* path, + int serializer(TqGroupHandle*, void**), + const void* deserializer(const void*, TqGroupHandle*), + void deleter(void*)) { //concat data file name and index file name - int fileFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755); + size_t pathLen = strlen(path); + char name[pathLen+10]; + strcpy(name, path); + strcat(name, "/" TQ_META_NAME); + int fileFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); if(fileFd < 0) return NULL; TqMetaStore* pMeta = malloc(sizeof(TqMetaStore)); if(pMeta == NULL) { @@ -35,7 +46,9 @@ TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*), memset(pMeta, 0, sizeof(TqMetaStore)); pMeta->fileFd = fileFd; - int idxFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755); + strcpy(name, path); + strcat(name, "/" TQ_IDX_NAME); + int idxFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); if(idxFd < 0) { //close file //free memory @@ -56,8 +69,29 @@ TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*), int32_t tqStoreClose(TqMetaStore* pMeta) { //commit data and idx - //close file + tqStorePersist(pMeta); + ASSERT(pMeta->unpersistHead && pMeta->unpersistHead->next==NULL); + close(pMeta->fileFd); + close(pMeta->idxFd); //free memory + for(int i = 0; i < TQ_BUCKET_SIZE; i++) { + TqMetaList* node = 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); + } + if(node->handle.valueInUse) { + pMeta->deleter(node->handle.valueInUse); + } + TqMetaList* next = node->next; + free(node); + node = next; + } + } + free(pMeta); return 0; } @@ -69,44 +103,202 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) { } int32_t tqStorePersist(TqMetaStore* pMeta) { - TqMetaList *node = pMeta->unpersistHead; - while(node->unpersistNext) { + int64_t idxBuf[3]; + TqMetaList *pHead = pMeta->unpersistHead; + TqMetaList *pNode = pHead->unpersistNext; + while(pHead != pNode) { + ASSERT(pNode->handle.valueInUse != NULL); //serialize + void* pBytes = NULL; + int sz = pMeta->serializer(pNode->handle.valueInUse, &pBytes); + ASSERT(pBytes != NULL); + //get current offset //append data - //write offset and idx + int nBytes = write(pMeta->fileFd, pBytes, sz); + //TODO: handle error in tfile + ASSERT(nBytes == sz); + + //write idx + //TODO: endian check and convert + idxBuf[0] = pNode->handle.key; + idxBuf[1] = pNode->handle.offset; + idxBuf[2] = (int64_t)sz; + nBytes = write(pMeta->idxFd, idxBuf, sizeof(idxBuf)); + //TODO: handle error in tfile + ASSERT(nBytes == sizeof(idxBuf)); + //remove from unpersist list + pHead->unpersistNext = pNode->unpersistNext; + pHead->unpersistNext->unpersistPrev = pHead; + + pNode->unpersistPrev = pNode->unpersistNext = NULL; + pNode = pHead->unpersistNext; + } + //TODO:fsync and return upper layer + return 0; +} + +int32_t tqHandlePutInUse(TqMetaStore* pMeta, int64_t key, void* value) { + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + //TODO: think about thread safety + pMeta->deleter(pNode->handle.valueInUse); + //change pointer ownership + pNode->handle.valueInUse = value; + } else { + pNode = pNode->next; + } } return 0; } -int32_t tqHandlePutInUse(TqMetaStore* pMeta, TqMetaHandle* handle) { - return 0; -} - TqMetaHandle* tqHandleGetInUse(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; + } else { + return NULL; + } + } else { + pNode = pNode->next; + } + } return NULL; } -int32_t tqHandlePutInTxn(TqMetaStore* pMeta, TqMetaHandle* handle) { +int32_t tqHandlePutInTxn(TqMetaStore* pMeta, int64_t key, void* value) { + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + //TODO: think about thread safety + pMeta->deleter(pNode->handle.valueInTxn); + //change pointer ownership + pNode->handle.valueInTxn = value; + } else { + pNode = pNode->next; + } + } return 0; } TqMetaHandle* tqHandleGetInTxn(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; + } else { + return NULL; + } + } else { + pNode = pNode->next; + } + } return NULL; } int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) { - return 0; + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + if(pNode->handle.valueInUse != NULL) { + pMeta->deleter(pNode->handle.valueInUse); + } + pNode->handle.valueInUse = pNode->handle.valueInTxn; + if(pNode->unpersistNext == NULL) { + pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; + pNode->unpersistPrev = pMeta->unpersistHead; + pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; + pMeta->unpersistHead->unpersistNext = pNode; + } + return 0; + } else { + pNode = pNode->next; + } + } + return -1; } int32_t tqHandleAbort(TqMetaStore* pMeta, int64_t key) { - return 0; + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + if(pNode->handle.valueInTxn != NULL) { + pMeta->deleter(pNode->handle.valueInTxn); + pNode->handle.valueInTxn = NULL; + return 0; + } + return -1; + } else { + pNode = pNode->next; + } + } + return -2; } int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) { - return 0; + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + while(pNode) { + if(pNode->handle.key == key) { + if(pNode->handle.valueInUse != NULL) { + pMeta->deleter(pNode->handle.valueInUse); + pNode->handle.valueInUse = NULL; + //if not in unpersist, put into unpersist + if(pNode->unpersistNext == NULL) { + pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; + pNode->unpersistPrev = pMeta->unpersistHead; + pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; + pMeta->unpersistHead->unpersistNext = pNode; + } + return 0; + } + return -1; + } else { + pNode = pNode->next; + } + } + return -2; } int32_t tqHandleClear(TqMetaStore* pMeta, int64_t key) { - return 0; + int64_t bucketKey = key & TQ_BUCKET_SIZE; + TqMetaList* pNode = pMeta->bucket[bucketKey]; + bool exist = false; + while(pNode) { + if(pNode->handle.key == key) { + if(pNode->handle.valueInUse != NULL) { + exist = true; + pMeta->deleter(pNode->handle.valueInUse); + pNode->handle.valueInUse = NULL; + } + if(pNode->handle.valueInTxn != NULL) { + exist = true; + pMeta->deleter(pNode->handle.valueInTxn); + pNode->handle.valueInTxn = NULL; + } + if(exist) { + if(pNode->unpersistNext == NULL) { + pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; + pNode->unpersistPrev = pMeta->unpersistHead; + pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; + pMeta->unpersistHead->unpersistNext = pNode; + } + return 0; + } + return -1; + } else { + pNode = pNode->next; + } + } + return -2; } From e42f02803462d050a4216b2c56cce3491bcbed4a Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 1 Nov 2021 16:45:48 +0800 Subject: [PATCH 2/3] refine tq meta interface --- source/server/vnode/tq/inc/tqMetaStore.h | 16 ++++++---------- source/server/vnode/tq/src/tqMetaStore.c | 12 ++++++++---- 2 files changed, 14 insertions(+), 14 deletions(-) diff --git a/source/server/vnode/tq/inc/tqMetaStore.h b/source/server/vnode/tq/inc/tqMetaStore.h index a82bf2a4a9..977980b2ae 100644 --- a/source/server/vnode/tq/inc/tqMetaStore.h +++ b/source/server/vnode/tq/inc/tqMetaStore.h @@ -61,19 +61,15 @@ int32_t tqStoreClose(TqMetaStore*); //int32_t TqStoreCommitAll(TqMetaStore*); int32_t tqStorePersist(TqMetaStore*); -TqMetaHandle* tqHandleGetInUse(TqMetaStore*, int64_t key); -int32_t tqHandlePutInUse(TqMetaStore*, int64_t key, void* value); -TqMetaHandle* tqHandleGetInTxn(TqMetaStore*, int64_t key); -int32_t tqHandlePutInTxn(TqMetaStore*, int64_t key, void* value); -//will replace old handle -//int32_t tqHandlePut(TqMetaStore*, TqMetaHandle* handle); -//delete in-use-handle, and put it in use +TqMetaHandle* tqHandleGet(TqMetaStore*, int64_t key); +int32_t tqHandlePut(TqMetaStore*, int64_t key, void* value); +//do commit int32_t tqHandleCommit(TqMetaStore*, int64_t key); -//delete in-txn-handle +//delete uncommitted int32_t tqHandleAbort(TqMetaStore*, int64_t key); -//delete in-use-handle +//delete committed int32_t tqHandleDel(TqMetaStore*, int64_t key); -//delete in-use-handle and in-txn-handle +//delete both committed and uncommitted int32_t tqHandleClear(TqMetaStore*, int64_t key); #ifdef __cplusplus diff --git a/source/server/vnode/tq/src/tqMetaStore.c b/source/server/vnode/tq/src/tqMetaStore.c index a93db7851b..90c2ad824b 100644 --- a/source/server/vnode/tq/src/tqMetaStore.c +++ b/source/server/vnode/tq/src/tqMetaStore.c @@ -22,6 +22,10 @@ #define TQ_META_NAME "tq.meta" #define TQ_IDX_NAME "tq.idx" + +static int32_t tqHandlePutCommitted(TqMetaStore*, int64_t key, void* value); +static TqMetaHandle* tqHandleGetUncommitted(TqMetaStore*, int64_t key); + typedef struct TqMetaPageBuf { int16_t offset; char buffer[TQ_PAGE_SIZE]; @@ -138,7 +142,7 @@ int32_t tqStorePersist(TqMetaStore* pMeta) { return 0; } -int32_t tqHandlePutInUse(TqMetaStore* pMeta, int64_t key, void* value) { +static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { @@ -154,7 +158,7 @@ int32_t tqHandlePutInUse(TqMetaStore* pMeta, int64_t key, void* value) { return 0; } -TqMetaHandle* tqHandleGetInUse(TqMetaStore* pMeta, int64_t key) { +TqMetaHandle* tqHandleGet(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { @@ -171,7 +175,7 @@ TqMetaHandle* tqHandleGetInUse(TqMetaStore* pMeta, int64_t key) { return NULL; } -int32_t tqHandlePutInTxn(TqMetaStore* pMeta, int64_t key, void* value) { +int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { @@ -187,7 +191,7 @@ int32_t tqHandlePutInTxn(TqMetaStore* pMeta, int64_t key, void* value) { return 0; } -TqMetaHandle* tqHandleGetInTxn(TqMetaStore* pMeta, int64_t key) { +static TqMetaHandle* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) { int64_t bucketKey = key & TQ_BUCKET_SIZE; TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { From f36d487e7b91bcfe8d3b36ae28d5222142b2d642 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 2 Nov 2021 17:27:19 +0800 Subject: [PATCH 3/3] implement tq meta file store --- include/server/vnode/tq/tq.h | 3 + source/server/vnode/tq/inc/tqMetaStore.h | 18 ++- source/server/vnode/tq/src/tq.c | 10 +- source/server/vnode/tq/src/tqMetaStore.c | 152 +++++++++++++++++------ 4 files changed, 140 insertions(+), 43 deletions(-) diff --git a/include/server/vnode/tq/tq.h b/include/server/vnode/tq/tq.h index 3aeaf9acb6..4a93491056 100644 --- a/include/server/vnode/tq/tq.h +++ b/include/server/vnode/tq/tq.h @@ -19,6 +19,9 @@ #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/server/vnode/tq/inc/tqMetaStore.h b/source/server/vnode/tq/inc/tqMetaStore.h index 977980b2ae..65429f7af2 100644 --- a/source/server/vnode/tq/inc/tqMetaStore.h +++ b/source/server/vnode/tq/inc/tqMetaStore.h @@ -20,6 +20,17 @@ #include "tq.h" #define TQ_BUCKET_SIZE 0xFF +#define TQ_PAGE_SIZE 4096 +//key + offset + size +#define TQ_IDX_ENTRY_SIZE 24 + +inline static int TqMaxEntryOnePage() { //170 + return TQ_PAGE_SIZE / TQ_IDX_ENTRY_SIZE; +} + +inline static int TqEmptyTail() { //16 + return TQ_PAGE_SIZE - TqMaxEntryOnePage(); +} #ifdef __cplusplus extern "C" { @@ -28,8 +39,9 @@ extern "C" { typedef struct TqMetaHandle { int64_t key; int64_t offset; - void *valueInUse; - void *valueInTxn; + int64_t serializedSize; + void* valueInUse; + void* valueInTxn; } TqMetaHandle; typedef struct TqMetaList { @@ -43,7 +55,7 @@ typedef struct TqMetaList { typedef struct TqMetaStore { TqMetaList* bucket[TQ_BUCKET_SIZE]; - //a table head, key is empty + //a table head TqMetaList* unpersistHead; int fileFd; //TODO:temporaral use, to be replaced by unified tfile int idxFd; //TODO:temporaral use, to be replaced by unified tfile diff --git a/source/server/vnode/tq/src/tq.c b/source/server/vnode/tq/src/tq.c index e8076515a7..1aa8f231c3 100644 --- a/source/server/vnode/tq/src/tq.c +++ b/source/server/vnode/tq/src/tq.c @@ -26,12 +26,12 @@ static int tqProtoCheck(TmqMsgHead *pMsg) { return pMsg->protoVer == 0; } -static int tqAckOneTopic(TqBufferHandle *bhandle, TmqOneAck *pAck, TqQueryMsg** ppQuery) { +static int tqAckOneTopic(TqBufferHandle *bHandle, TmqOneAck *pAck, TqQueryMsg** ppQuery) { //clean old item and move forward int32_t consumeOffset = pAck->consumeOffset; int idx = consumeOffset % TQ_BUFFER_SIZE; - ASSERT(bhandle->buffer[idx].content && bhandle->buffer[idx].executor); - tfree(bhandle->buffer[idx].content); + ASSERT(bHandle->buffer[idx].content && bHandle->buffer[idx].executor); + tfree(bHandle->buffer[idx].content); if( 1 /* TODO: need to launch new query */) { TqQueryMsg* pNewQuery = malloc(sizeof(TqQueryMsg)); if(pNewQuery == NULL) { @@ -39,10 +39,10 @@ static int tqAckOneTopic(TqBufferHandle *bhandle, TmqOneAck *pAck, TqQueryMsg** return -1; } //TODO: lock executor - pNewQuery->exec->executor = bhandle->buffer[idx].executor; + pNewQuery->exec->executor = bHandle->buffer[idx].executor; //TODO: read from wal and assign to src pNewQuery->exec->src = 0; - pNewQuery->exec->dest = &bhandle->buffer[idx]; + pNewQuery->exec->dest = &bHandle->buffer[idx]; pNewQuery->next = *ppQuery; *ppQuery = pNewQuery; } diff --git a/source/server/vnode/tq/src/tqMetaStore.c b/source/server/vnode/tq/src/tqMetaStore.c index 90c2ad824b..72edf10a1f 100644 --- a/source/server/vnode/tq/src/tqMetaStore.c +++ b/source/server/vnode/tq/src/tqMetaStore.c @@ -13,12 +13,11 @@ * along with this program. If not, see . */ #include "tqMetaStore.h" -//TODO:replace by a abstract file layer +//TODO:replace by an abstract file layer #include #include #include -#define TQ_PAGE_SIZE 4096 #define TQ_META_NAME "tq.meta" #define TQ_IDX_NAME "tq.idx" @@ -35,21 +34,16 @@ TqMetaStore* tqStoreOpen(const char* path, int serializer(TqGroupHandle*, void**), const void* deserializer(const void*, TqGroupHandle*), void deleter(void*)) { - //concat data file name and index file name - size_t pathLen = strlen(path); - char name[pathLen+10]; - strcpy(name, path); - strcat(name, "/" TQ_META_NAME); - int fileFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); - if(fileFd < 0) return NULL; TqMetaStore* pMeta = malloc(sizeof(TqMetaStore)); if(pMeta == NULL) { //close return NULL; } - memset(pMeta, 0, sizeof(TqMetaStore)); - pMeta->fileFd = fileFd; - + + //concat data file name and index file name + size_t pathLen = strlen(path); + char name[pathLen+10]; + strcpy(name, path); strcat(name, "/" TQ_IDX_NAME); int idxFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); @@ -58,6 +52,7 @@ TqMetaStore* tqStoreOpen(const char* path, //free memory return NULL; } + pMeta->idxFd = idxFd; pMeta->unpersistHead = malloc(sizeof(TqMetaList)); if(pMeta->unpersistHead == NULL) { @@ -65,9 +60,37 @@ TqMetaStore* tqStoreOpen(const char* path, //free memory return NULL; } + + strcpy(name, path); + strcat(name, "/" TQ_META_NAME); + int fileFd = open(name, O_WRONLY | O_CREAT | O_EXCL, 0755); + if(fileFd < 0) return NULL; + + memset(pMeta, 0, sizeof(TqMetaStore)); + pMeta->fileFd = fileFd; + pMeta->serializer = serializer; pMeta->deserializer = deserializer; 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) { + //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)); + if(pNode == NULL) { + //TODO: free memory and return error + } + memcpy(&pNode->handle, &readBuf[i], TQ_IDX_ENTRY_SIZE); + int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE; + pNode->next = pMeta->bucket[bucketKey]; + pMeta->bucket[bucketKey] = pNode; + } + } + return pMeta; } @@ -106,30 +129,66 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) { return 0; } +//TODO: wrap in tfile int32_t tqStorePersist(TqMetaStore* pMeta) { - int64_t idxBuf[3]; + char writeBuf[TQ_PAGE_SIZE]; + int64_t* bufPtr = (int64_t*)writeBuf; TqMetaList *pHead = pMeta->unpersistHead; TqMetaList *pNode = pHead->unpersistNext; while(pHead != pNode) { - ASSERT(pNode->handle.valueInUse != NULL); + 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)); + + //remove from list + int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE; + TqMetaList* pBucketHead = pMeta->bucket[bucketKey]; + if(pBucketHead == pNode) { + pMeta->bucket[bucketKey] = pBucketHead->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); + } + 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 - idxBuf[0] = pNode->handle.key; - idxBuf[1] = pNode->handle.offset; - idxBuf[2] = (int64_t)sz; - nBytes = write(pMeta->idxFd, idxBuf, sizeof(idxBuf)); - //TODO: handle error in tfile - ASSERT(nBytes == sizeof(idxBuf)); + *(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 pHead->unpersistNext = pNode->unpersistNext; @@ -138,7 +197,16 @@ int32_t tqStorePersist(TqMetaStore* pMeta) { pNode->unpersistPrev = pNode->unpersistNext = NULL; pNode = pHead->unpersistNext; } - //TODO:fsync and return upper layer + //write left bytes + if((char*)bufPtr != writeBuf) { + int used = (char*)bufPtr - writeBuf; + int nBytes = write(pMeta->idxFd, writeBuf, used); + //TODO: handle error in tfile + ASSERT(nBytes == used); + } + //TODO: using fsync in tfile + fsync(pMeta->idxFd); + fsync(pMeta->fileFd); return 0; } @@ -151,10 +219,24 @@ static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value pMeta->deleter(pNode->handle.valueInUse); //change pointer ownership pNode->handle.valueInUse = value; + 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.valueInUse = value; + //put into unpersist list + pNewNode->unpersistPrev = pMeta->unpersistHead; + pNewNode->unpersistNext = pMeta->unpersistHead->unpersistNext; + pMeta->unpersistHead->unpersistNext->unpersistPrev = pNewNode; + pMeta->unpersistHead->unpersistNext = pNewNode; return 0; } @@ -184,10 +266,19 @@ int32_t tqHandlePut(TqMetaStore* pMeta, int64_t key, void* value) { pMeta->deleter(pNode->handle.valueInTxn); //change pointer ownership pNode->handle.valueInTxn = value; + 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 = value; return 0; } @@ -254,24 +345,15 @@ int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) { TqMetaList* pNode = pMeta->bucket[bucketKey]; while(pNode) { if(pNode->handle.key == key) { - if(pNode->handle.valueInUse != NULL) { - pMeta->deleter(pNode->handle.valueInUse); - pNode->handle.valueInUse = NULL; - //if not in unpersist, put into unpersist - if(pNode->unpersistNext == NULL) { - pNode->unpersistNext = pMeta->unpersistHead->unpersistNext; - pNode->unpersistPrev = pMeta->unpersistHead; - pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode; - pMeta->unpersistHead->unpersistNext = pNode; - } - return 0; - } - return -1; + pMeta->deleter(pNode->handle.valueInTxn); + pNode->handle.valueInTxn = NULL; + return 0; } else { pNode = pNode->next; } } - return -2; + //no such key + return -1; } int32_t tqHandleClear(TqMetaStore* pMeta, int64_t key) {