Merge branch 'feature/tq' into 3.0

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

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})

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,9 +37,23 @@ inline static int TqEmptyTail() { //16
   return TQ_PAGE_SIZE - TqMaxEntryOnePage();
 }
 
-#ifdef __cplusplus
-extern "C" {
-#endif
+#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;
@@ -59,30 +78,33 @@ 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*);
+  char* dirPath;
+  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(TqGroupHandle*, void**),
-    const void* deserializer(const void*, TqGroupHandle*),
-    void deleter(void*));
+    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*);
 //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 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);
+int32_t tqHandleCommit(TqMetaStore*, int64_t key);
 //delete uncommitted
-int32_t       tqHandleAbort(TqMetaStore*, int64_t key);
-//delete committed
-int32_t       tqHandleDel(TqMetaStore*, int64_t key);
+int32_t tqHandleAbort(TqMetaStore*, int64_t key);
+//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);
+int32_t tqHandleClear(TqMetaStore*, int64_t key);
 
 #ifdef __cplusplus
 }

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 <fcntl.h>
 #include <string.h>
 #include <unistd.h>
@@ -22,8 +23,18 @@
 #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);
+
+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;
@@ -31,23 +42,34 @@ 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, TqSerializedHead** ppHead),
+    const void* deserializer(const TqSerializedHead* pHead, 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);
+  pMeta->dirPath = malloc(pathLen+1);
+  if(pMeta->dirPath != NULL) {
+    //TODO: memory insufficient
+  }
+  strcpy(pMeta->dirPath, 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 +78,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 +103,103 @@ 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];
+  TqSerializedHead* serializedObj = malloc(TQ_PAGE_SIZE);
+  if(serializedObj == NULL) {
+    //TODO:memory insufficient
+  }
+  int idxRead;
+  int allocated = TQ_PAGE_SIZE;
+  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(allocated < pNode->handle.serializedSize) {
+        void *ptr = realloc(serializedObj, pNode->handle.serializedSize);
+        if(ptr == NULL) {
+          //TODO: memory insufficient
+        }
+        serializedObj = ptr;
+        allocated = pNode->handle.serializedSize;
+      }
+      serializedObj->ssize = pNode->handle.serializedSize;
+      if(read(fileFd, serializedObj, pNode->handle.serializedSize) != pNode->handle.serializedSize) {
+        //TODO: read error
+      }
+      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(serializedObj);
   return pMeta;
 }
 
@@ -102,30 +211,54 @@ int32_t tqStoreClose(TqMetaStore* pMeta) {
   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);
+    TqMetaList* pNode = pMeta->bucket[i];
+    while(pNode) {
+      ASSERT(pNode->unpersistNext == NULL);
+      ASSERT(pNode->unpersistPrev == NULL);
+      if(pNode->handle.valueInTxn
+          && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
+        pMeta->deleter(pNode->handle.valueInTxn);
       }
-      if(node->handle.valueInUse) {
-        pMeta->deleter(node->handle.valueInUse);
+      if(pNode->handle.valueInUse
+          && pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
+        pMeta->deleter(pNode->handle.valueInUse);
       }
-      TqMetaList* next = node->next;
-      free(node);
-      node = next;
+      TqMetaList* next = pNode->next;
+      free(pNode);
+      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
+          && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
+        pMeta->deleter(pNode->handle.valueInTxn);
+      }
+      if(pNode->handle.valueInUse
+          && pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
+        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;
 }
 
@@ -135,69 +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
-      int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE;
-      TqMetaList* pBucketHead = pMeta->bucket[bucketKey];
-      if(pBucketHead == pNode) {
-        pMeta->bucket[bucketKey] = pBucketHead->next;
+    if(pNode->handle.valueInUse) {
+      if(pNode->handle.valueInTxn) {
+        pSHead->action = TQ_ACTION_INUSE_CONT;
       } 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);
-        }
+        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);
     }
-    //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;
+    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
+    //write idx file
     //TODO: endian check and convert
     *(bufPtr++) = pNode->handle.key;
     *(bufPtr++) = pNode->handle.offset;
-    *(bufPtr++) = (int64_t)sz;
+    *(bufPtr++) = (int64_t)nBytes;
     if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) {
       nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf));
-      //TODO: handle error in tfile
+      //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] = pNode->next;
+      } else {
+        TqMetaList* pBucketNode = pBucketHead;
+        while(pBucketNode->next != NULL
+            && pBucketNode->next != pNode) {
+          pBucketNode = pBucketNode->next; 
+        }
+        //impossible for pBucket->next == NULL
+        ASSERT(pBucketNode->next == pNode);
+        pBucketNode->next = pNode->next;
+      }
+      free(pNode);
+    }
   }
+
   //write left bytes
+  free(pSHead);
   if((char*)bufPtr != writeBuf) {
     int used = (char*)bufPtr - writeBuf;
     int nBytes = write(pMeta->idxFd, writeBuf, used);
@@ -216,7 +369,10 @@ 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
+          && pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
+        pMeta->deleter(pNode->handle.valueInUse);
+      }
       //change pointer ownership
       pNode->handle.valueInUse = value;
       return 0;
@@ -240,13 +396,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;
       }
@@ -257,15 +413,19 @@ TqMetaHandle* 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) {
     if(pNode->handle.key == key) {
       //TODO: think about thread safety
-      pMeta->deleter(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;
@@ -279,16 +439,58 @@ 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;
+  tqLinkUnpersist(pMeta, pNewNode);
   return 0;
 }
 
-static TqMetaHandle* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) {
+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) {
-      if(pNode->handle.valueInTxn != NULL) {
-        return &pNode->handle;
+      //TODO: think about thread safety
+      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;
+    }
+  }
+  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;
+  tqLinkUnpersist(pMeta, pNewNode);
+  return 0;
+}
+
+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
+          && pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
+        return pNode->handle.valueInTxn;
       } else {
         return NULL;
       }
@@ -304,16 +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 != NULL) {
+      if(pNode->handle.valueInUse
+          && pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
         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;
-      }
+      pNode->handle.valueInTxn = NULL;
+      tqLinkUnpersist(pMeta, pNode);
       return 0;
     } else {
       pNode = pNode->next;
@@ -327,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;
@@ -344,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;
@@ -364,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;

source/dnode/vnode/tq/test/CMakeLists.txt

+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
+)

source/dnode/vnode/tq/test/tqMetaTest.cpp

+#include <gtest/gtest.h>
+#include <cstring>
+#include <iostream>
+#include <queue>
+
+#include "tqMetaStore.h"
+
+struct Foo {
+  int32_t a;
+};
+
+int FooSerializer(const void* pObj, TqSerializedHead** ppHead) {
+  Foo* foo = (Foo*) pObj;
+  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 TqSerializedHead* pHead, void** ppObj) {
+  if(*ppObj == NULL) {
+    *ppObj = realloc(*ppObj, sizeof(int32_t));
+  }
+  Foo* pFoo = *(Foo**)ppObj;
+  pFoo->a = *(int32_t*)pHead->content; 
+  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, 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;
+  tqHandleMovePut(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;
+  tqHandleMovePut(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;
+  tqHandleMovePut(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);
+}
+
+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);
+}

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
+#endif
\ No newline at end of file