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

refactor tq

This commit is contained in:
Liu Jicong 2021-12-17 17:44:36 +08:00
parent d037b24e05
commit 0e69034454
7 changed files with 296 additions and 284 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/taosmsg.h
View File

@ -50,7 +50,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONSUME, "mq-consume" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_QUERY, "mq-query" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_QUERY, "mq-query" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_DISCONNECT, "mq-disconnect" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_DISCONNECT, "mq-disconnect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_SET, "mq-set" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_SET_CUR, "mq-set-cur" )
// message from client to mnode // message from client to mnode
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONNECT, "connect" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONNECT, "connect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_ACCT, "create-acct" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_ACCT, "create-acct" )

171
include/dnode/vnode/tq/tq.h
View File

@ -18,84 +18,71 @@
#include "mallocator.h" #include "mallocator.h"
#include "os.h" #include "os.h"
#include "tlist.h"
#include "tutil.h" #include "tutil.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
typedef struct TmqMsgHead { typedef struct STqMsgHead {
int32_t protoVer; int32_t protoVer;
int32_t msgType; int32_t msgType;
int64_t cgId; int64_t cgId;
int64_t clientId; int64_t clientId;
} TmqMsgHead; } STqMsgHead;
typedef struct TmqOneAck { typedef struct STqOneAck {
int64_t topicId; int64_t topicId;
int64_t consumeOffset; int64_t consumeOffset;
} TmqOneAck; } STqOneAck;
typedef struct TmqAcks { typedef struct STqAcks {
int32_t ackNum; int32_t ackNum;
// should be sorted // should be sorted
TmqOneAck acks[]; STqOneAck acks[];
} TmqAcks; } STqAcks;
// TODO: put msgs into common typedef struct STqSetCurReq {
typedef struct TmqConnectReq { STqMsgHead head;
TmqMsgHead head; int64_t topicId;
TmqAcks acks; int64_t offset;
} TmqConnectReq; } STqSetCurReq;
typedef struct TmqConnectRsp {
TmqMsgHead head;
int8_t status;
} TmqConnectRsp;
typedef struct TmqDisconnectReq {
TmqMsgHead head;
} TmqDiscconectReq;
typedef struct TmqDisconnectRsp {
TmqMsgHead head;
int8_t status;
} TmqDisconnectRsp;
typedef struct STqConsumeReq { typedef struct STqConsumeReq {
TmqMsgHead head; STqMsgHead head;
TmqAcks acks; STqAcks acks;
} STqConsumeReq; } STqConsumeReq;
typedef struct TmqMsgContent { typedef struct STqMsgContent {
int64_t topicId; int64_t topicId;
int64_t msgLen; int64_t msgLen;
char msg[]; char msg[];
} TmqMsgContent; } STqMsgContent;
typedef struct STqConsumeRsp { typedef struct STqConsumeRsp {
TmqMsgHead head; STqMsgHead head;
int64_t bodySize; int64_t bodySize;
TmqMsgContent msgs[]; STqMsgContent msgs[];
} STqConsumeRsp; } STqConsumeRsp;
typedef struct TmqSubscribeReq { typedef struct STqSubscribeReq {
TmqMsgHead head; STqMsgHead head;
int32_t topicNum; int32_t topicNum;
int64_t topic[]; int64_t topic[];
} TmqSubscribeReq; } STqSubscribeReq;
typedef struct tmqSubscribeRsp { typedef struct STqSubscribeRsp {
TmqMsgHead head; STqMsgHead head;
int64_t vgId; int64_t vgId;
char ep[TSDB_EP_LEN]; // TSDB_EP_LEN char ep[TSDB_EP_LEN]; // TSDB_EP_LEN
} TmqSubscribeRsp; } STqSubscribeRsp;
typedef struct TmqHeartbeatReq { typedef struct STqHeartbeatReq {
} TmqHeartbeatReq; } STqHeartbeatReq;
typedef struct TmqHeartbeatRsp { typedef struct STqHeartbeatRsp {
} TmqHeartbeatRsp; } STqHeartbeatRsp;
typedef struct STqTopicVhandle { typedef struct STqTopicVhandle {
int64_t topicId; int64_t topicId;
@ -113,39 +100,41 @@ typedef struct STqBufferItem {
// executors are identical but not concurrent // executors are identical but not concurrent
// so there must be a copy in each item // so there must be a copy in each item
void* executor; void* executor;
int32_t status;
int64_t size; int64_t size;
void* content; void* content;
} STqBufferItem; } STqMsgItem;
typedef struct STqBufferHandle { typedef struct STqTopic {
// char* topic; //c style, end with '\0' // char* topic; //c style, end with '\0'
// int64_t cgId; // int64_t cgId;
// void* ahandle; // void* ahandle;
int64_t nextConsumeOffset; int64_t nextConsumeOffset;
int64_t floatingCursor; int64_t floatingCursor;
int64_t topicId; int64_t topicId;
int32_t head; int32_t head;
int32_t tail; int32_t tail;
STqBufferItem buffer[TQ_BUFFER_SIZE]; STqMsgItem buffer[TQ_BUFFER_SIZE];
} STqBufferHandle; } STqTopic;
typedef struct STqListHandle { typedef struct STqListHandle {
STqBufferHandle bufHandle; STqTopic topic;
struct STqListHandle* next; struct STqListHandle* next;
} STqListHandle; } STqList;
typedef struct STqGroupHandle { typedef struct STqGroup {
int64_t cId; int64_t cId;
int64_t cgId; int64_t cgId;
void* ahandle; void* ahandle;
int32_t topicNum; int32_t topicNum;
STqListHandle* head; STqList* head;
} STqGroupHandle; SList* topicList; // SList<STqTopic>
} STqGroup;
typedef struct STqQueryExec { typedef struct STqQueryExec {
void* src; void* src;
STqBufferItem* dest; STqMsgItem* dest;
void* executor; void* executor;
} STqQueryExec; } STqQueryExec;
typedef struct STqQueryMsg { typedef struct STqQueryMsg {
@ -209,15 +198,15 @@ typedef void (*FTqDelete)(void*);
#define TQ_DUP_INTXN_REWRITE 0 #define TQ_DUP_INTXN_REWRITE 0
#define TQ_DUP_INTXN_REJECT 2 #define TQ_DUP_INTXN_REJECT 2
static inline bool TqUpdateAppend(int32_t tqConfigFlag) { return tqConfigFlag & TQ_UPDATE_APPEND; } static inline bool tqUpdateAppend(int32_t tqConfigFlag) { return tqConfigFlag & TQ_UPDATE_APPEND; }
static inline bool TqDupIntxnReject(int32_t tqConfigFlag) { return tqConfigFlag & TQ_DUP_INTXN_REJECT; } static inline bool tqDupIntxnReject(int32_t tqConfigFlag) { return tqConfigFlag & TQ_DUP_INTXN_REJECT; }
static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST; static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST;
#define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE #define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE
typedef struct TqMetaHandle { typedef struct STqMetaHandle {
int64_t key; int64_t key;
int64_t offset; int64_t offset;
int64_t serializedSize; int64_t serializedSize;
@ -225,23 +214,25 @@ typedef struct TqMetaHandle {
void* valueInTxn; void* valueInTxn;
} STqMetaHandle; } STqMetaHandle;
typedef struct TqMetaList { typedef struct STqMetaList {
STqMetaHandle handle; STqMetaHandle handle;
struct TqMetaList* next; struct STqMetaList* next;
// struct TqMetaList* inTxnPrev; // struct STqMetaList* inTxnPrev;
// struct TqMetaList* inTxnNext; // struct STqMetaList* inTxnNext;
struct TqMetaList* unpersistPrev; struct STqMetaList* unpersistPrev;
struct TqMetaList* unpersistNext; struct STqMetaList* unpersistNext;
} STqMetaList; } STqMetaList;
typedef struct TqMetaStore { typedef struct STqMetaStore {
STqMetaList* bucket[TQ_BUCKET_SIZE]; STqMetaList* bucket[TQ_BUCKET_SIZE];
// a table head // a table head
STqMetaList* unpersistHead; STqMetaList* unpersistHead;
// TODO:temporaral use, to be replaced by unified tfile // TODO:temporaral use, to be replaced by unified tfile
int fileFd; int fileFd;
// TODO:temporaral use, to be replaced by unified tfile // TODO:temporaral use, to be replaced by unified tfile
int idxFd; int idxFd;
char* dirPath; char* dirPath;
int32_t tqConfigFlag; int32_t tqConfigFlag;
FTqSerialize pSerializer; FTqSerialize pSerializer;
@ -250,8 +241,8 @@ typedef struct TqMetaStore {
} STqMetaStore; } STqMetaStore;
typedef struct STQ { typedef struct STQ {
// the collection of group handle // the collection of groups
// the handle of kvstore // the handle of meta kvstore
char* path; char* path;
STqCfg* tqConfig; STqCfg* tqConfig;
STqLogReader* tqLogReader; STqLogReader* tqLogReader;
@ -266,23 +257,25 @@ void tqClose(STQ*);
// void* will be replace by a msg type // void* will be replace by a msg type
int tqPushMsg(STQ*, void* msg, int64_t version); int tqPushMsg(STQ*, void* msg, int64_t version);
int tqCommit(STQ*); int tqCommit(STQ*);
int tqSetCursor(STQ*, void* msg);
int tqConsume(STQ*, STqConsumeReq*); int tqConsume(STQ*, STqConsumeReq*);
STqGroupHandle* tqGetGroupHandle(STQ*, int64_t cId); int tqSetCursor(STQ*, STqSetCurReq* pMsg);
int tqBufferSetOffset(STqTopic*, int64_t offset);
STqGroupHandle* tqOpenTCGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId); STqTopic* tqFindTopic(STqGroup*, int64_t topicId);
int tqCloseTCGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId);
int tqMoveOffsetToNext(STqGroupHandle*);
int tqResetOffset(STQ*, int64_t topicId, int64_t cgId, int64_t offset);
int tqRegisterContext(STqGroupHandle*, void* ahandle);
int tqLaunchQuery(STqGroupHandle*);
int tqSendLaunchQuery(STqGroupHandle*);
int tqSerializeGroupHandle(const STqGroupHandle* gHandle, STqSerializedHead** ppHead); STqGroup* tqGetGroup(STQ*, int64_t clientId);
const void* tqDeserializeGroupHandle(const STqSerializedHead* pHead, STqGroupHandle** gHandle); STqGroup* tqOpenGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId);
int tqCloseGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId);
int tqRegisterContext(STqGroup*, void* ahandle);
int tqSendLaunchQuery(STqMsgItem*, int64_t offset);
int tqSerializeGroup(const STqGroup*, STqSerializedHead**);
const void* tqDeserializeGroup(const STqSerializedHead*, STqGroup**);
static int tqQueryExecuting(int32_t status) { return status; }
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -44,7 +44,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONSUME] = dndProcessVnodeQueryMsg; pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONSUME] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONNECT] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_MQ_CONNECT] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_MQ_DISCONNECT] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_MQ_DISCONNECT] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_MQ_SET] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_MQ_SET_CUR] = dndProcessVnodeWriteMsg;
// msg from client to mnode // msg from client to mnode
pMgmt->msgFp[TSDB_MSG_TYPE_CONNECT] = dndProcessMnodeReadMsg; pMgmt->msgFp[TSDB_MSG_TYPE_CONNECT] = dndProcessMnodeReadMsg;

6
source/dnode/vnode/impl/src/vnodeWrite.c
View File

@ -16,17 +16,13 @@
#include "vnodeDef.h" #include "vnodeDef.h"
int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) { int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) {
SVnodeReq *pVnodeReq;
switch (pMsg->msgType) { switch (pMsg->msgType) {
case TSDB_MSG_TYPE_MQ_SET: case TSDB_MSG_TYPE_MQ_SET_CUR:
if (tqSetCursor(pVnode->pTq, pMsg->pCont) < 0) { if (tqSetCursor(pVnode->pTq, pMsg->pCont) < 0) {
// TODO: handle error // TODO: handle error
} }
break; break;
} }
void *pBuf = pMsg->pCont;
return 0; return 0;
} }

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

@ -24,21 +24,16 @@
// handle management message // handle management message
// //
int tqGetgHandleSSize(const STqGroupHandle* gHandle); int tqGroupSSize(const STqGroup* pGroup);
int tqBufHandleSSize(); int tqTopicSSize();
int tqBufItemSSize(); int tqItemSSize();
STqGroupHandle* tqFindHandle(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) { void* tqSerializeListHandle(STqList* listHandle, void* ptr);
STqGroupHandle* gHandle; void* tqSerializeTopic(STqTopic* pTopic, void* ptr);
return NULL; void* tqSerializeItem(STqMsgItem* pItem, void* ptr);
}
void* tqSerializeListHandle(STqListHandle* listHandle, void* ptr); const void* tqDeserializeTopic(const void* pBytes, STqTopic* pTopic);
void* tqSerializeBufHandle(STqBufferHandle* bufHandle, void* ptr); const void* tqDeserializeItem(const void* pBytes, STqMsgItem* pItem);
void* tqSerializeBufItem(STqBufferItem* bufItem, void* ptr);
const void* tqDeserializeBufHandle(const void* pBytes, STqBufferHandle* bufHandle);
const void* tqDeserializeBufItem(const void* pBytes, STqBufferItem* bufItem);
STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemAllocatorFactory* allocFac) { STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemAllocatorFactory* allocFac) {
STQ* pTq = malloc(sizeof(STQ)); STQ* pTq = malloc(sizeof(STQ));
@ -54,8 +49,7 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemA
if (pTq->tqMemRef.pAllocator == NULL) { if (pTq->tqMemRef.pAllocator == NULL) {
// TODO // TODO
} }
pTq->tqMeta = pTq->tqMeta = tqStoreOpen(path, (FTqSerialize)tqSerializeGroup, (FTqDeserialize)tqDeserializeGroup, free, 0);
tqStoreOpen(path, (FTqSerialize)tqSerializeGroupHandle, (FTqDeserialize)tqDeserializeGroupHandle, free, 0);
if (pTq->tqMeta == NULL) { if (pTq->tqMeta == NULL) {
// TODO: free STQ // TODO: free STQ
return NULL; return NULL;
@ -63,14 +57,14 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemA
return pTq; return pTq;
} }
static int tqProtoCheck(TmqMsgHead* pMsg) { return pMsg->protoVer == 0; } static int tqProtoCheck(STqMsgHead* pMsg) { return pMsg->protoVer == 0; }
static int tqAckOneTopic(STqBufferHandle* bHandle, TmqOneAck* pAck, STqQueryMsg** ppQuery) { static int tqAckOneTopic(STqTopic* pTopic, STqOneAck* pAck, STqQueryMsg** ppQuery) {
// clean old item and move forward // clean old item and move forward
int32_t consumeOffset = pAck->consumeOffset; int32_t consumeOffset = pAck->consumeOffset;
int idx = consumeOffset % TQ_BUFFER_SIZE; int idx = consumeOffset % TQ_BUFFER_SIZE;
ASSERT(bHandle->buffer[idx].content && bHandle->buffer[idx].executor); ASSERT(pTopic->buffer[idx].content && pTopic->buffer[idx].executor);
tfree(bHandle->buffer[idx].content); tfree(pTopic->buffer[idx].content);
if (1 /* TODO: need to launch new query */) { if (1 /* TODO: need to launch new query */) {
STqQueryMsg* pNewQuery = malloc(sizeof(STqQueryMsg)); STqQueryMsg* pNewQuery = malloc(sizeof(STqQueryMsg));
if (pNewQuery == NULL) { if (pNewQuery == NULL) {
@ -78,29 +72,29 @@ static int tqAckOneTopic(STqBufferHandle* bHandle, TmqOneAck* pAck, STqQueryMsg*
return -1; return -1;
} }
// TODO: lock executor // TODO: lock executor
pNewQuery->exec->executor = bHandle->buffer[idx].executor; pNewQuery->exec->executor = pTopic->buffer[idx].executor;
// TODO: read from wal and assign to src // TODO: read from wal and assign to src
pNewQuery->exec->src = 0; pNewQuery->exec->src = 0;
pNewQuery->exec->dest = &bHandle->buffer[idx]; pNewQuery->exec->dest = &pTopic->buffer[idx];
pNewQuery->next = *ppQuery; pNewQuery->next = *ppQuery;
*ppQuery = pNewQuery; *ppQuery = pNewQuery;
} }
return 0; return 0;
} }
static int tqAck(STqGroupHandle* gHandle, TmqAcks* pAcks) { static int tqAck(STqGroup* pGroup, STqAcks* pAcks) {
int32_t ackNum = pAcks->ackNum; int32_t ackNum = pAcks->ackNum;
TmqOneAck* acks = pAcks->acks; STqOneAck* acks = pAcks->acks;
// double ptr for acks and list // double ptr for acks and list
int i = 0; int i = 0;
STqListHandle* node = gHandle->head; STqList* node = pGroup->head;
int ackCnt = 0; int ackCnt = 0;
STqQueryMsg* pQuery = NULL; STqQueryMsg* pQuery = NULL;
while (i < ackNum && node->next) { while (i < ackNum && node->next) {
if (acks[i].topicId == node->next->bufHandle.topicId) { if (acks[i].topicId == node->next->topic.topicId) {
ackCnt++; ackCnt++;
tqAckOneTopic(&node->next->bufHandle, &acks[i], &pQuery); tqAckOneTopic(&node->next->topic, &acks[i], &pQuery);
} else if (acks[i].topicId < node->next->bufHandle.topicId) { } else if (acks[i].topicId < node->next->topic.topicId) {
i++; i++;
} else { } else {
node = node->next; node = node->next;
@ -112,28 +106,29 @@ static int tqAck(STqGroupHandle* gHandle, TmqAcks* pAcks) {
return ackCnt; return ackCnt;
} }
static int tqCommitTCGroup(STqGroupHandle* handle) { static int tqCommitGroup(STqGroup* pGroup) {
// persist modification into disk // persist modification into disk
return 0; return 0;
} }
int tqCreateTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId, STqGroupHandle** handle) { int tqCreateGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId, STqGroup** ppGroup) {
// create in disk // create in disk
STqGroupHandle* gHandle = (STqGroupHandle*)malloc(sizeof(STqGroupHandle)); STqGroup* pGroup = (STqGroup*)malloc(sizeof(STqGroup));
if (gHandle == NULL) { if (pGroup == NULL) {
// TODO // TODO
return -1; return -1;
} }
memset(gHandle, 0, sizeof(STqGroupHandle)); *ppGroup = pGroup;
memset(pGroup, 0, sizeof(STqGroup));
return 0; return 0;
} }
STqGroupHandle* tqOpenTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) { STqGroup* tqOpenGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
STqGroupHandle* gHandle = tqHandleGet(pTq->tqMeta, cId); STqGroup* pGroup = tqHandleGet(pTq->tqMeta, cId);
if (gHandle == NULL) { if (pGroup == NULL) {
int code = tqCreateTCGroup(pTq, topicId, cgId, cId, &gHandle); int code = tqCreateGroup(pTq, topicId, cgId, cId, &pGroup);
if (code != 0) { if (code < 0) {
// TODO // TODO
return NULL; return NULL;
} }
@ -141,23 +136,26 @@ STqGroupHandle* tqOpenTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t c
// create // create
// open // open
return gHandle; return pGroup;
} }
int tqCloseTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) { return 0; } int tqCloseGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
// TODO
return 0;
}
int tqDropTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) { int tqDropGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
// delete from disk // delete from disk
return 0; return 0;
} }
static int tqFetch(STqGroupHandle* gHandle, void** msg) { static int tqFetch(STqGroup* pGroup, void** msg) {
STqListHandle* head = gHandle->head; STqList* head = pGroup->head;
STqListHandle* node = head; STqList* node = head;
int totSize = 0; int totSize = 0;
// TODO: make it a macro // TODO: make it a macro
int sizeLimit = 4 * 1024; int sizeLimit = 4 * 1024;
TmqMsgContent* buffer = malloc(sizeLimit); STqMsgContent* buffer = malloc(sizeLimit);
if (buffer == NULL) { if (buffer == NULL) {
// TODO:memory insufficient // TODO:memory insufficient
return -1; return -1;
@ -166,25 +164,25 @@ static int tqFetch(STqGroupHandle* gHandle, void** msg) {
// until all topic iterated or msgs over sizeLimit // until all topic iterated or msgs over sizeLimit
while (node->next) { while (node->next) {
node = node->next; node = node->next;
STqBufferHandle* bufHandle = &node->bufHandle; STqTopic* topicHandle = &node->topic;
int idx = bufHandle->nextConsumeOffset % TQ_BUFFER_SIZE; int idx = topicHandle->nextConsumeOffset % TQ_BUFFER_SIZE;
if (bufHandle->buffer[idx].content != NULL && bufHandle->buffer[idx].offset == bufHandle->nextConsumeOffset) { if (topicHandle->buffer[idx].content != NULL && topicHandle->buffer[idx].offset == topicHandle->nextConsumeOffset) {
totSize += bufHandle->buffer[idx].size; totSize += topicHandle->buffer[idx].size;
if (totSize > sizeLimit) { if (totSize > sizeLimit) {
void* ptr = realloc(buffer, totSize); void* ptr = realloc(buffer, totSize);
if (ptr == NULL) { if (ptr == NULL) {
totSize -= bufHandle->buffer[idx].size; totSize -= topicHandle->buffer[idx].size;
// TODO:memory insufficient // TODO:memory insufficient
// return msgs already copied // return msgs already copied
break; break;
} }
} }
*((int64_t*)buffer) = bufHandle->topicId; *((int64_t*)buffer) = topicHandle->topicId;
buffer = POINTER_SHIFT(buffer, sizeof(int64_t)); buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
*((int64_t*)buffer) = bufHandle->buffer[idx].size; *((int64_t*)buffer) = topicHandle->buffer[idx].size;
buffer = POINTER_SHIFT(buffer, sizeof(int64_t)); buffer = POINTER_SHIFT(buffer, sizeof(int64_t));
memcpy(buffer, bufHandle->buffer[idx].content, bufHandle->buffer[idx].size); memcpy(buffer, topicHandle->buffer[idx].content, topicHandle->buffer[idx].size);
buffer = POINTER_SHIFT(buffer, bufHandle->buffer[idx].size); buffer = POINTER_SHIFT(buffer, topicHandle->buffer[idx].size);
if (totSize > sizeLimit) { if (totSize > sizeLimit) {
break; break;
} }
@ -193,11 +191,19 @@ static int tqFetch(STqGroupHandle* gHandle, void** msg) {
return totSize; return totSize;
} }
STqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) { return NULL; } STqGroup* tqGetGroup(STQ* pTq, int64_t clientId) { return tqHandleGet(pTq->tqMeta, clientId); }
int tqLaunchQuery(STqGroupHandle* gHandle) { return 0; } int tqSendLaunchQuery(STqMsgItem* bufItem, int64_t offset) {
if (tqQueryExecuting(bufItem->status)) {
int tqSendLaunchQuery(STqGroupHandle* gHandle) { return 0; } return 0;
}
bufItem->status = 1;
// load data from wal or buffer pool
// put into exec
// send exec into non blocking queue
// when query finished, put into buffer pool
return 0;
}
/*int tqMoveOffsetToNext(TqGroupHandle* gHandle) {*/ /*int tqMoveOffsetToNext(TqGroupHandle* gHandle) {*/
/*return 0;*/ /*return 0;*/
@ -214,23 +220,69 @@ int tqCommit(STQ* pTq) {
return 0; return 0;
} }
int tqSetCursor(STQ* pTq, void* msg) { int tqBufferSetOffset(STqTopic* pTopic, int64_t offset) {
int code;
memset(pTopic->buffer, 0, sizeof(pTopic->buffer));
// launch query
for (int i = offset; i < offset + TQ_BUFFER_SIZE; i++) {
int pos = i % TQ_BUFFER_SIZE;
code = tqSendLaunchQuery(&pTopic->buffer[pos], offset);
if (code < 0) {
// TODO: error handling
}
}
// set offset
pTopic->nextConsumeOffset = offset;
pTopic->floatingCursor = offset;
return 0; return 0;
} }
int tqConsume(STQ* pTq, STqConsumeReq* pMsg) { STqTopic* tqFindTopic(STqGroup* pGroup, int64_t topicId) {
if (!tqProtoCheck((TmqMsgHead*)pMsg)) { // TODO
// proto version invalid return NULL;
return -1; }
}
int64_t clientId = pMsg->head.clientId; int tqSetCursor(STQ* pTq, STqSetCurReq* pMsg) {
STqGroupHandle* gHandle = tqGetGroupHandle(pTq, clientId); int code;
int64_t clientId = pMsg->head.clientId;
int64_t topicId = pMsg->topicId;
int64_t offset = pMsg->offset;
STqGroup* gHandle = tqGetGroup(pTq, clientId);
if (gHandle == NULL) { if (gHandle == NULL) {
// client not connect // client not connect
return -1; return -1;
} }
STqTopic* topicHandle = tqFindTopic(gHandle, topicId);
if (topicHandle == NULL) {
return -1;
}
if (pMsg->offset == topicHandle->nextConsumeOffset) {
return 0;
}
// TODO: check log last version
code = tqBufferSetOffset(topicHandle, offset);
if (code < 0) {
// set error code
return -1;
}
return 0;
}
int tqConsume(STQ* pTq, STqConsumeReq* pMsg) {
if (!tqProtoCheck((STqMsgHead*)pMsg)) {
// proto version invalid
return -1;
}
int64_t clientId = pMsg->head.clientId;
STqGroup* pGroup = tqGetGroup(pTq, clientId);
if (pGroup == NULL) {
// client not connect
return -1;
}
if (pMsg->acks.ackNum != 0) { if (pMsg->acks.ackNum != 0) {
if (tqAck(gHandle, &pMsg->acks) != 0) { if (tqAck(pGroup, &pMsg->acks) != 0) {
// ack not success // ack not success
return -1; return -1;
} }
@ -238,22 +290,22 @@ int tqConsume(STQ* pTq, STqConsumeReq* pMsg) {
STqConsumeRsp* pRsp = (STqConsumeRsp*)pMsg; STqConsumeRsp* pRsp = (STqConsumeRsp*)pMsg;
if (tqFetch(gHandle, (void**)&pRsp->msgs) <= 0) { if (tqFetch(pGroup, (void**)&pRsp->msgs) <= 0) {
// fetch error // fetch error
return -1; return -1;
} }
// judge and launch new query // judge and launch new query
if (tqLaunchQuery(gHandle)) { /*if (tqSendLaunchQuery(gHandle)) {*/
// launch query error // launch query error
return -1; /*return -1;*/
} /*}*/
return 0; return 0;
} }
int tqSerializeGroupHandle(const STqGroupHandle* gHandle, STqSerializedHead** ppHead) { int tqSerializeGroup(const STqGroup* pGroup, STqSerializedHead** ppHead) {
// calculate size // calculate size
int sz = tqGetgHandleSSize(gHandle) + sizeof(STqSerializedHead); int sz = tqGroupSSize(pGroup) + sizeof(STqSerializedHead);
if (sz > (*ppHead)->ssize) { if (sz > (*ppHead)->ssize) {
void* tmpPtr = realloc(*ppHead, sz); void* tmpPtr = realloc(*ppHead, sz);
if (tmpPtr == NULL) { if (tmpPtr == NULL) {
@ -266,52 +318,52 @@ int tqSerializeGroupHandle(const STqGroupHandle* gHandle, STqSerializedHead** pp
} }
void* ptr = (*ppHead)->content; void* ptr = (*ppHead)->content;
// do serialization // do serialization
*(int64_t*)ptr = gHandle->cId; *(int64_t*)ptr = pGroup->cId;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
*(int64_t*)ptr = gHandle->cgId; *(int64_t*)ptr = pGroup->cgId;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
*(int32_t*)ptr = gHandle->topicNum; *(int32_t*)ptr = pGroup->topicNum;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
if (gHandle->topicNum > 0) { if (pGroup->topicNum > 0) {
tqSerializeListHandle(gHandle->head, ptr); tqSerializeListHandle(pGroup->head, ptr);
} }
return 0; return 0;
} }
void* tqSerializeListHandle(STqListHandle* listHandle, void* ptr) { void* tqSerializeListHandle(STqList* listHandle, void* ptr) {
STqListHandle* node = listHandle; STqList* node = listHandle;
ASSERT(node != NULL); ASSERT(node != NULL);
while (node) { while (node) {
ptr = tqSerializeBufHandle(&node->bufHandle, ptr); ptr = tqSerializeTopic(&node->topic, ptr);
node = node->next; node = node->next;
} }
return ptr; return ptr;
} }
void* tqSerializeBufHandle(STqBufferHandle* bufHandle, void* ptr) { void* tqSerializeTopic(STqTopic* pTopic, void* ptr) {
*(int64_t*)ptr = bufHandle->nextConsumeOffset; *(int64_t*)ptr = pTopic->nextConsumeOffset;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
*(int64_t*)ptr = bufHandle->topicId; *(int64_t*)ptr = pTopic->topicId;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
*(int32_t*)ptr = bufHandle->head; *(int32_t*)ptr = pTopic->head;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
*(int32_t*)ptr = bufHandle->tail; *(int32_t*)ptr = pTopic->tail;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
for (int i = 0; i < TQ_BUFFER_SIZE; i++) { for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
ptr = tqSerializeBufItem(&bufHandle->buffer[i], ptr); ptr = tqSerializeItem(&pTopic->buffer[i], ptr);
} }
return ptr; return ptr;
} }
void* tqSerializeBufItem(STqBufferItem* bufItem, void* ptr) { void* tqSerializeItem(STqMsgItem* bufItem, void* ptr) {
// TODO: do we need serialize this? // TODO: do we need serialize this?
// mainly for executor // mainly for executor
return ptr; return ptr;
} }
const void* tqDeserializeGroupHandle(const STqSerializedHead* pHead, STqGroupHandle** ppGHandle) { const void* tqDeserializeGroup(const STqSerializedHead* pHead, STqGroup** ppGroup) {
STqGroupHandle* gHandle = *ppGHandle; STqGroup* gHandle = *ppGroup;
const void* ptr = pHead->content; const void* ptr = pHead->content;
gHandle->cId = *(int64_t*)ptr; gHandle->cId = *(int64_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
gHandle->cgId = *(int64_t*)ptr; gHandle->cgId = *(int64_t*)ptr;
@ -320,63 +372,63 @@ const void* tqDeserializeGroupHandle(const STqSerializedHead* pHead, STqGroupHan
gHandle->topicNum = *(int32_t*)ptr; gHandle->topicNum = *(int32_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
gHandle->head = NULL; gHandle->head = NULL;
STqListHandle* node = gHandle->head; STqList* node = gHandle->head;
for (int i = 0; i < gHandle->topicNum; i++) { for (int i = 0; i < gHandle->topicNum; i++) {
if (gHandle->head == NULL) { if (gHandle->head == NULL) {
if ((node = malloc(sizeof(STqListHandle))) == NULL) { if ((node = malloc(sizeof(STqList))) == NULL) {
// TODO: error // TODO: error
return NULL; return NULL;
} }
node->next = NULL; node->next = NULL;
ptr = tqDeserializeBufHandle(ptr, &node->bufHandle); ptr = tqDeserializeTopic(ptr, &node->topic);
gHandle->head = node; gHandle->head = node;
} else { } else {
node->next = malloc(sizeof(STqListHandle)); node->next = malloc(sizeof(STqList));
if (node->next == NULL) { if (node->next == NULL) {
// TODO: error // TODO: error
return NULL; return NULL;
} }
node->next->next = NULL; node->next->next = NULL;
ptr = tqDeserializeBufHandle(ptr, &node->next->bufHandle); ptr = tqDeserializeTopic(ptr, &node->next->topic);
node = node->next; node = node->next;
} }
} }
return ptr; return ptr;
} }
const void* tqDeserializeBufHandle(const void* pBytes, STqBufferHandle* bufHandle) { const void* tqDeserializeTopic(const void* pBytes, STqTopic* topic) {
const void* ptr = pBytes; const void* ptr = pBytes;
bufHandle->nextConsumeOffset = *(int64_t*)ptr; topic->nextConsumeOffset = *(int64_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
bufHandle->topicId = *(int64_t*)ptr; topic->topicId = *(int64_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int64_t)); ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
bufHandle->head = *(int32_t*)ptr; topic->head = *(int32_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
bufHandle->tail = *(int32_t*)ptr; topic->tail = *(int32_t*)ptr;
ptr = POINTER_SHIFT(ptr, sizeof(int32_t)); ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
for (int i = 0; i < TQ_BUFFER_SIZE; i++) { for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
ptr = tqDeserializeBufItem(ptr, &bufHandle->buffer[i]); ptr = tqDeserializeItem(ptr, &topic->buffer[i]);
} }
return ptr; return ptr;
} }
const void* tqDeserializeBufItem(const void* pBytes, STqBufferItem* bufItem) { return pBytes; } const void* tqDeserializeItem(const void* pBytes, STqMsgItem* bufItem) { return pBytes; }
// TODO: make this a macro // TODO: make this a macro
int tqGetgHandleSSize(const STqGroupHandle* gHandle) { int tqGroupSSize(const STqGroup* gHandle) {
return sizeof(int64_t) * 2 // cId + cgId return sizeof(int64_t) * 2 // cId + cgId
+ sizeof(int32_t) // topicNum + sizeof(int32_t) // topicNum
+ gHandle->topicNum * tqBufHandleSSize(); + gHandle->topicNum * tqTopicSSize();
} }
// TODO: make this a macro // TODO: make this a macro
int tqBufHandleSSize() { int tqTopicSSize() {
return sizeof(int64_t) * 2 // nextConsumeOffset + topicId return sizeof(int64_t) * 2 // nextConsumeOffset + topicId
+ sizeof(int32_t) * 2 // head + tail + sizeof(int32_t) * 2 // head + tail
+ TQ_BUFFER_SIZE * tqBufItemSSize(); + TQ_BUFFER_SIZE * tqItemSSize();
} }
int tqBufItemSSize() { int tqItemSSize() {
// TODO: do this need serialization? // TODO: do this need serialization?
// mainly for executor // mainly for executor
return 0; return 0;

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

@ -472,7 +472,7 @@ static inline int32_t tqHandlePutImpl(STqMetaStore* pMeta, int64_t key, void* va
if (pNode->handle.key == key) { if (pNode->handle.key == key) {
// TODO: think about thread safety // TODO: think about thread safety
if (pNode->handle.valueInTxn) { if (pNode->handle.valueInTxn) {
if (TqDupIntxnReject(pMeta->tqConfigFlag)) { if (tqDupIntxnReject(pMeta->tqConfigFlag)) {
return -2; return -2;
} }
if (pNode->handle.valueInTxn != TQ_DELETE_TOKEN) { if (pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {

121
source/dnode/vnode/tq/test/tqMetaTest.cpp
View File

@ -10,8 +10,8 @@ struct Foo {
}; };
int FooSerializer(const void* pObj, STqSerializedHead** ppHead) { int FooSerializer(const void* pObj, STqSerializedHead** ppHead) {
Foo* foo = (Foo*) pObj; Foo* foo = (Foo*)pObj;
if((*ppHead) == NULL || (*ppHead)->ssize < sizeof(STqSerializedHead) + sizeof(int32_t)) { if ((*ppHead) == NULL || (*ppHead)->ssize < sizeof(STqSerializedHead) + sizeof(int32_t)) {
*ppHead = (STqSerializedHead*)realloc(*ppHead, sizeof(STqSerializedHead) + sizeof(int32_t)); *ppHead = (STqSerializedHead*)realloc(*ppHead, sizeof(STqSerializedHead) + sizeof(int32_t));
(*ppHead)->ssize = sizeof(STqSerializedHead) + sizeof(int32_t); (*ppHead)->ssize = sizeof(STqSerializedHead) + sizeof(int32_t);
} }
@ -20,36 +20,28 @@ int FooSerializer(const void* pObj, STqSerializedHead** ppHead) {
} }
const void* FooDeserializer(const STqSerializedHead* pHead, void** ppObj) { const void* FooDeserializer(const STqSerializedHead* pHead, void** ppObj) {
if(*ppObj == NULL) { if (*ppObj == NULL) {
*ppObj = realloc(*ppObj, sizeof(int32_t)); *ppObj = realloc(*ppObj, sizeof(int32_t));
} }
Foo* pFoo = *(Foo**)ppObj; Foo* pFoo = *(Foo**)ppObj;
pFoo->a = *(int32_t*)pHead->content; pFoo->a = *(int32_t*)pHead->content;
return NULL; return NULL;
} }
void FooDeleter(void* pObj) { void FooDeleter(void* pObj) { free(pObj); }
free(pObj);
}
class TqMetaUpdateAppendTest : public ::testing::Test { class TqMetaUpdateAppendTest : public ::testing::Test {
protected: protected:
void SetUp() override {
taosRemoveDir(pathName);
pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
ASSERT(pMeta);
}
void SetUp() override { void TearDown() override { tqStoreClose(pMeta); }
taosRemoveDir(pathName);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta);
}
void TearDown() override { STqMetaStore* pMeta;
tqStoreClose(pMeta); const char* pathName = "/tmp/tq_test";
}
TqMetaStore* pMeta;
const char* pathName = "/tmp/tq_test";
}; };
TEST_F(TqMetaUpdateAppendTest, copyPutTest) { TEST_F(TqMetaUpdateAppendTest, copyPutTest) {
@ -57,11 +49,11 @@ TEST_F(TqMetaUpdateAppendTest, copyPutTest) {
foo.a = 3; foo.a = 3;
tqHandleCopyPut(pMeta, 1, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, 1, &foo, sizeof(Foo));
Foo* pFoo = (Foo*) tqHandleGet(pMeta, 1); Foo* pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
tqHandleCommit(pMeta, 1); tqHandleCommit(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo->a, 3); EXPECT_EQ(pFoo->a, 3);
} }
@ -78,10 +70,7 @@ TEST_F(TqMetaUpdateAppendTest, persistTest) {
EXPECT_EQ(pBar == NULL, true); EXPECT_EQ(pBar == NULL, true);
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
pBar = (Foo*)tqHandleGet(pMeta, 1); pBar = (Foo*)tqHandleGet(pMeta, 1);
@ -97,7 +86,7 @@ TEST_F(TqMetaUpdateAppendTest, uncommittedTest) {
pFoo->a = 3; pFoo->a = 3;
tqHandleMovePut(pMeta, 1, pFoo); tqHandleMovePut(pMeta, 1, pFoo);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
} }
@ -106,11 +95,11 @@ TEST_F(TqMetaUpdateAppendTest, abortTest) {
pFoo->a = 3; pFoo->a = 3;
tqHandleMovePut(pMeta, 1, pFoo); tqHandleMovePut(pMeta, 1, pFoo);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
tqHandleAbort(pMeta, 1); tqHandleAbort(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
} }
@ -119,32 +108,29 @@ TEST_F(TqMetaUpdateAppendTest, deleteTest) {
pFoo->a = 3; pFoo->a = 3;
tqHandleMovePut(pMeta, 1, pFoo); tqHandleMovePut(pMeta, 1, pFoo);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
tqHandleCommit(pMeta, 1); tqHandleCommit(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
ASSERT_EQ(pFoo != NULL, true); ASSERT_EQ(pFoo != NULL, true);
EXPECT_EQ(pFoo->a, 3); EXPECT_EQ(pFoo->a, 3);
tqHandleDel(pMeta, 1); tqHandleDel(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
ASSERT_EQ(pFoo != NULL, true); ASSERT_EQ(pFoo != NULL, true);
EXPECT_EQ(pFoo->a, 3); EXPECT_EQ(pFoo->a, 3);
tqHandleCommit(pMeta, 1); tqHandleCommit(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
pFoo = (Foo*) tqHandleGet(pMeta, 1); pFoo = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true); EXPECT_EQ(pFoo == NULL, true);
} }
@ -162,10 +148,7 @@ TEST_F(TqMetaUpdateAppendTest, intxnPersist) {
EXPECT_EQ(pFoo1->a, 3); EXPECT_EQ(pFoo1->a, 3);
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
pFoo1 = (Foo*)tqHandleGet(pMeta, 1); pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
@ -177,10 +160,7 @@ TEST_F(TqMetaUpdateAppendTest, intxnPersist) {
EXPECT_EQ(pFoo1->a, 4); EXPECT_EQ(pFoo1->a, 4);
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
pFoo1 = (Foo*)tqHandleGet(pMeta, 1); pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
@ -190,13 +170,13 @@ TEST_F(TqMetaUpdateAppendTest, intxnPersist) {
TEST_F(TqMetaUpdateAppendTest, multiplePage) { TEST_F(TqMetaUpdateAppendTest, multiplePage) {
srand(0); srand(0);
std::vector<int> v; std::vector<int> v;
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
v.push_back(rand()); v.push_back(rand());
Foo foo; Foo foo;
foo.a = v[i]; foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
} }
for(int i = 0; i < 500; i++) { for (int i = 0; i < 500; i++) {
tqHandleCommit(pMeta, i); tqHandleCommit(pMeta, i);
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i); Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n"; ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
@ -204,38 +184,34 @@ TEST_F(TqMetaUpdateAppendTest, multiplePage) {
} }
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
for(int i = 500; i < 1000; i++) { for (int i = 500; i < 1000; i++) {
tqHandleCommit(pMeta, i); tqHandleCommit(pMeta, i);
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i); Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n"; ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]); EXPECT_EQ(pFoo->a, v[i]);
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i); Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n"; ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]); EXPECT_EQ(pFoo->a, v[i]);
} }
} }
TEST_F(TqMetaUpdateAppendTest, multipleRewrite) { TEST_F(TqMetaUpdateAppendTest, multipleRewrite) {
srand(0); srand(0);
std::vector<int> v; std::vector<int> v;
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
v.push_back(rand()); v.push_back(rand());
Foo foo; Foo foo;
foo.a = v[i]; foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
} }
for(int i = 0; i < 500; i++) { for (int i = 0; i < 500; i++) {
tqHandleCommit(pMeta, i); tqHandleCommit(pMeta, i);
v[i] = rand(); v[i] = rand();
Foo foo; Foo foo;
@ -243,25 +219,22 @@ TEST_F(TqMetaUpdateAppendTest, multipleRewrite) {
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
} }
for(int i = 500; i < 1000; i++) { for (int i = 500; i < 1000; i++) {
v[i] = rand(); v[i] = rand();
Foo foo; Foo foo;
foo.a = v[i]; foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
tqHandleCommit(pMeta, i); tqHandleCommit(pMeta, i);
} }
tqStoreClose(pMeta); tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName, pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
FooSerializer, FooDeserializer, FooDeleter,
TQ_UPDATE_APPEND
);
ASSERT(pMeta); ASSERT(pMeta);
for(int i = 500; i < 1000; i++) { for (int i = 500; i < 1000; i++) {
v[i] = rand(); v[i] = rand();
Foo foo; Foo foo;
foo.a = v[i]; foo.a = v[i];
@ -269,40 +242,38 @@ TEST_F(TqMetaUpdateAppendTest, multipleRewrite) {
tqHandleCommit(pMeta, i); tqHandleCommit(pMeta, i);
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i); Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n"; ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]); EXPECT_EQ(pFoo->a, v[i]);
} }
} }
TEST_F(TqMetaUpdateAppendTest, dupCommit) { TEST_F(TqMetaUpdateAppendTest, dupCommit) {
srand(0); srand(0);
std::vector<int> v; std::vector<int> v;
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
v.push_back(rand()); v.push_back(rand());
Foo foo; Foo foo;
foo.a = v[i]; foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo)); tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
int ret = tqHandleCommit(pMeta, i); int ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, 0); EXPECT_EQ(ret, 0);
ret = tqHandleCommit(pMeta, i); ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, -1); EXPECT_EQ(ret, -1);
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
int ret = tqHandleCommit(pMeta, i); int ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, -1); EXPECT_EQ(ret, -1);
} }
for(int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i); Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n"; ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]); EXPECT_EQ(pFoo->a, v[i]);
} }
} }