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