xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/tq' into 3.0

This commit is contained in:
Liu Jicong 2021-11-02 17:34:50 +08:00
parent d0a4c4ad6b f36d487e7b
commit 06f55a36d4
4 changed files with 355 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/server/vnode/tq/tq.h
View File

@ -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

55
source/dnode/vnode/tq/inc/tqMetaStore.h
View File

@ -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

36
source/dnode/vnode/tq/src/tq.c
View File

@ -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();

322
source/dnode/vnode/tq/src/tqMetaStore.c
View File

@ -13,34 +13,46 @@
* 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_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;
}