diff --git a/include/server/vnode/tq/tq.h b/include/server/vnode/tq/tq.h
index 3aeaf9acb6be607caea6bf60d8627875fa952df8..4a9349105637eaa52dd41379557eb84a5f6120e7 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/dnode/vnode/tq/inc/tqMetaStore.h b/source/dnode/vnode/tq/inc/tqMetaStore.h
index 066e76028d43d0fdacf3971863ba0b83a04d119b..65429f7af29369b3f4192629c6b165a08cb0d3ab 100644
--- a/source/dnode/vnode/tq/inc/tqMetaStore.h
+++ b/source/dnode/vnode/tq/inc/tqMetaStore.h
@@ -17,9 +17,20 @@
#define _TQ_META_STORE_H_
#include "os.h"
+#include "tq.h"
-#define TQ_INUSE_SIZE 0xFF
+#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,47 +39,49 @@ 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 {
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];
- //a table head, key is empty
+ TqMetaList* bucket[TQ_BUCKET_SIZE];
+ //a table head
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);
-TqMetaHandle* tqHandleGetInTxn(TqMetaStore*, int64_t key);
-int32_t tqHandlePutInTxn(TqMetaStore*, TqMetaHandle* handle);
-//delete in-use-handle, make in-txn-handle 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/dnode/vnode/tq/src/tq.c b/source/dnode/vnode/tq/src/tq.c
index 7ecdfe7f197a2d22f3e00d9757f2899b7469289f..1aa8f231c34dc0b110475444037d69756fe30859 100644
--- a/source/dnode/vnode/tq/src/tq.c
+++ b/source/dnode/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,22 +39,22 @@ 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;
}
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/dnode/vnode/tq/src/tqMetaStore.c b/source/dnode/vnode/tq/src/tqMetaStore.c
index d652058c7452b92b3a0f803221f4782d56671fa6..72edf10a1f8be4d41ca4c32ee15618b1402a9cd8 100644
--- a/source/dnode/vnode/tq/src/tqMetaStore.c
+++ b/source/dnode/vnode/tq/src/tqMetaStore.c
@@ -13,34 +13,46 @@
* 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_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];
} TqMetaPageBuf;
-TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*),
- void* deserializer(void*), void deleter(void*)) {
- //concat data file name and index file name
- int fileFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755);
- if(fileFd < 0) return NULL;
+TqMetaStore* tqStoreOpen(const char* path,
+ int serializer(TqGroupHandle*, void**),
+ const void* deserializer(const void*, TqGroupHandle*),
+ void deleter(void*)) {
TqMetaStore* pMeta = malloc(sizeof(TqMetaStore));
if(pMeta == NULL) {
//close
return NULL;
}
- memset(pMeta, 0, sizeof(TqMetaStore));
- pMeta->fileFd = fileFd;
-
- int idxFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755);
+
+ //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);
if(idxFd < 0) {
//close file
//free memory
return NULL;
}
+
pMeta->idxFd = idxFd;
pMeta->unpersistHead = malloc(sizeof(TqMetaList));
if(pMeta->unpersistHead == NULL) {
@@ -48,16 +60,65 @@ TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*),
//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;
}
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;
}
@@ -68,45 +129,262 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) {
return 0;
}
+//TODO: wrap in tfile
int32_t tqStorePersist(TqMetaStore* pMeta) {
- TqMetaList *node = pMeta->unpersistHead;
- while(node->unpersistNext) {
+ char writeBuf[TQ_PAGE_SIZE];
+ int64_t* bufPtr = (int64_t*)writeBuf;
+ TqMetaList *pHead = pMeta->unpersistHead;
+ TqMetaList *pNode = pHead->unpersistNext;
+ 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));
+
+ //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
- //write offset and idx
+ 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));
+ //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;
+ pHead->unpersistNext->unpersistPrev = pHead;
+
+ pNode->unpersistPrev = pNode->unpersistNext = NULL;
+ pNode = pHead->unpersistNext;
}
+ //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;
}
-int32_t tqHandlePutInUse(TqMetaStore* pMeta, TqMetaHandle* handle) {
+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) {
+ if(pNode->handle.key == key) {
+ //TODO: think about thread safety
+ 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;
}
-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) {
+ 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 tqHandlePut(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;
+ 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;
}
-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) {
+ 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) {
+ pMeta->deleter(pNode->handle.valueInTxn);
+ pNode->handle.valueInTxn = NULL;
+ return 0;
+ } else {
+ pNode = pNode->next;
+ }
+ }
+ //no such key
+ return -1;
}
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;
}