implement tq meta file store

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

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

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

source/server/vnode/tq/src/tqMetaStore.c

@@ -13,12 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 #include "tqMetaStore.h"
-//TODO:replace by a abstract file layer
+//TODO:replace by an abstract file layer
 #include <fcntl.h>
 #include <string.h>
 #include <unistd.h>
 
-#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) {