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

Merge branch '3.0' into feature/vnode

This commit is contained in:
Hongze Cheng 2021-11-10 10:03:05 +08:00
parent ef4f731454 771310d322
commit b1df458777
11 changed files with 311 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
include/libs/sync/sync.h
View File

@ -29,10 +29,10 @@ typedef int64_t SyncIndex;
typedef uint64_t SyncTerm;
typedef enum {
TAOS_SYNC_ROLE_FOLLOWER = 0,
TAOS_SYNC_ROLE_CANDIDATE = 1,
TAOS_SYNC_ROLE_LEADER = 2,
} ESyncRole;
TAOS_SYNC_STATE_FOLLOWER = 0,
TAOS_SYNC_STATE_CANDIDATE = 1,
TAOS_SYNC_STATE_LEADER = 2,
} ESyncState;
typedef struct {
void* data;
@ -55,7 +55,7 @@ typedef struct {
int32_t selfIndex;
int32_t replica;
SNodeInfo node[TSDB_MAX_REPLICA];
ESyncRole role[TSDB_MAX_REPLICA];
ESyncState role[TSDB_MAX_REPLICA];
} SNodesRole;
typedef struct SSyncFSM {
@ -159,9 +159,9 @@ void syncStop(const SSyncNode*);
int32_t syncPropose(SSyncNode* syncNode, const SSyncBuffer* pBuf, void* pData, bool isWeak);
// int32_t syncAddNode(SSyncNode syncNode, const SNodeInfo *pNode);
int32_t syncAddNode(SSyncNode syncNode, const SNodeInfo *pNode);
// int32_t syncRemoveNode(SSyncNode syncNode, const SNodeInfo *pNode);
int32_t syncRemoveNode(SSyncNode syncNode, const SNodeInfo *pNode);
extern int32_t sDebugFlag;

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

@ -24,18 +24,18 @@
extern "C" {
#endif
#define TQ_BUCKET_SIZE 0xFF
#define TQ_BUCKET_MASK 0xFF
#define TQ_BUCKET_SIZE 256
#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();
}
#define TQ_IDX_SIZE 24
//4096 / 24
#define TQ_MAX_IDX_ONE_PAGE 170
//24 * 170
#define TQ_IDX_PAGE_BODY_SIZE 4080
//4096 - 4080
#define TQ_IDX_PAGE_HEAD_SIZE 16
#define TQ_ACTION_CONST 0
#define TQ_ACTION_INUSE 1
@ -92,19 +92,17 @@ int32_t tqStoreClose(TqMetaStore*);
//int32_t tqStoreDelete(TqMetaStore*);
//int32_t TqStoreCommitAll(TqMetaStore*);
int32_t tqStorePersist(TqMetaStore*);
//clean deleted idx and data from persistent file
int32_t tqStoreCompact(TqMetaStore*);
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);
//delete uncommitted
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 tqHandleCommit(TqMetaStore*, int64_t key);
int32_t tqHandleAbort(TqMetaStore*, int64_t key);
#ifdef __cplusplus
}

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

@ -22,6 +22,9 @@
//
//handle management message
//
int tqGetgHandleSSize(const TqGroupHandle *gHandle);
static int tqProtoCheck(TmqMsgHead *pMsg) {
return pMsg->protoVer == 0;
}

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

@ -27,19 +27,46 @@ 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;
}
if(pNode->unpersistNext == NULL) {
pNode->unpersistNext = pMeta->unpersistHead->unpersistNext;
pNode->unpersistPrev = pMeta->unpersistHead;
pMeta->unpersistHead->unpersistNext->unpersistPrev = pNode;
pMeta->unpersistHead->unpersistNext = pNode;
}
}
static inline int tqSeekLastPage(int fd) {
int offset = lseek(fd, 0, SEEK_END);
int pageNo = offset / TQ_PAGE_SIZE;
int curPageOffset = pageNo * TQ_PAGE_SIZE;
return lseek(fd, curPageOffset, SEEK_SET);
}
typedef struct TqMetaPageBuf {
int16_t offset;
char buffer[TQ_PAGE_SIZE];
} TqMetaPageBuf;
//TODO: the struct is tightly coupled with index entry
typedef struct TqIdxPageHead {
int16_t writeOffset;
int8_t unused[14];
} TqIdxPageHead;
typedef struct TqIdxPageBuf {
TqIdxPageHead head;
char buffer[TQ_IDX_PAGE_BODY_SIZE];
} TqIdxPageBuf;
static inline int tqReadLastPage(int fd, TqIdxPageBuf* pBuf) {
int offset = tqSeekLastPage(fd);
int nBytes;
if((nBytes = read(fd, pBuf, TQ_PAGE_SIZE)) == -1) {
return -1;
}
if(nBytes == 0) {
memset(pBuf, 0, TQ_PAGE_SIZE);
pBuf->head.writeOffset = TQ_IDX_PAGE_HEAD_SIZE;
}
ASSERT(nBytes == 0 || nBytes == pBuf->head.writeOffset);
return lseek(fd, offset, SEEK_SET);
}
TqMetaStore* tqStoreOpen(const char* path,
int serializer(const void* pObj, TqSerializedHead** ppHead),
@ -103,27 +130,30 @@ TqMetaStore* tqStoreOpen(const char* path,
pMeta->deleter = deleter;
//read idx file and load into memory
char idxBuf[TQ_PAGE_SIZE];
TqIdxPageBuf idxBuf;
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))) {
bool readEnd = false;
while((idxRead = read(idxFd, &idxBuf, TQ_PAGE_SIZE))) {
if(idxRead == -1) {
//TODO: handle error
ASSERT(false);
}
ASSERT(idxBuf.head.writeOffset == idxRead);
//loop read every entry
for(int i = 0; i < idxRead; i += TQ_IDX_ENTRY_SIZE) {
for(int i = 0; i < idxBuf.head.writeOffset - TQ_IDX_PAGE_HEAD_SIZE; i += TQ_IDX_SIZE) {
TqMetaList *pNode = malloc(sizeof(TqMetaList));
if(pNode == NULL) {
//TODO: free memory and return error
}
memset(pNode, 0, sizeof(TqMetaList));
memcpy(&pNode->handle, &idxBuf[i], TQ_IDX_ENTRY_SIZE);
lseek(fileFd, pNode->handle.offset, SEEK_CUR);
memcpy(&pNode->handle, &idxBuf.buffer[i], TQ_IDX_SIZE);
lseek(fileFd, pNode->handle.offset, SEEK_SET);
if(allocated < pNode->handle.serializedSize) {
void *ptr = realloc(serializedObj, pNode->handle.serializedSize);
if(ptr == NULL) {
@ -154,9 +184,9 @@ TqMetaStore* tqStoreOpen(const char* path,
} else {
pNode->handle.valueInUse = TQ_DELETE_TOKEN;
}
serializedObj = POINTER_SHIFT(serializedObj, serializedObj->ssize);
if(serializedObj->ssize != sizeof(TqSerializedHead)) {
pMeta->deserializer(serializedObj, &pNode->handle.valueInTxn);
TqSerializedHead* ptr = POINTER_SHIFT(serializedObj, serializedObj->ssize);
if(ptr->ssize != sizeof(TqSerializedHead)) {
pMeta->deserializer(ptr, &pNode->handle.valueInTxn);
} else {
pNode->handle.valueInTxn = TQ_DELETE_TOKEN;
}
@ -165,7 +195,7 @@ TqMetaStore* tqStoreOpen(const char* path,
}
//put into list
int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE;
int bucketKey = pNode->handle.key & TQ_BUCKET_MASK;
TqMetaList* pBucketNode = pMeta->bucket[bucketKey];
if(pBucketNode == NULL) {
pMeta->bucket[bucketKey] = pNode;
@ -174,15 +204,18 @@ TqMetaStore* tqStoreOpen(const char* path,
pMeta->bucket[bucketKey] = pNode;
} else {
while(pBucketNode->next &&
pBucketNode->next->handle.key == pNode->handle.key) {
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;
TqMetaList *pNodeFound = pBucketNode->next;
pNode->next = pNodeFound->next;
pBucketNode->next = pNode;
pBucketNode = pNodeFound;
} else {
pNode->next = pMeta->bucket[bucketKey];
pMeta->bucket[bucketKey] = pNode;
pBucketNode = NULL;
}
}
@ -264,8 +297,8 @@ int32_t tqStoreDelete(TqMetaStore* pMeta) {
//TODO: wrap in tfile
int32_t tqStorePersist(TqMetaStore* pMeta) {
char writeBuf[TQ_PAGE_SIZE];
int64_t* bufPtr = (int64_t*)writeBuf;
TqIdxPageBuf idxBuf;
int64_t* bufPtr = (int64_t*)idxBuf.buffer;
TqMetaList *pHead = pMeta->unpersistHead;
TqMetaList *pNode = pHead->unpersistNext;
TqSerializedHead *pSHead = malloc(sizeof(TqSerializedHead));
@ -278,6 +311,17 @@ int32_t tqStorePersist(TqMetaStore* pMeta) {
pSHead->ssize = sizeof(TqSerializedHead);
int allocatedSize = sizeof(TqSerializedHead);
int offset = lseek(pMeta->fileFd, 0, SEEK_CUR);
tqReadLastPage(pMeta->idxFd, &idxBuf);
if(idxBuf.head.writeOffset == TQ_PAGE_SIZE) {
lseek(pMeta->idxFd, 0, SEEK_END);
memset(&idxBuf, 0, TQ_PAGE_SIZE);
idxBuf.head.writeOffset = TQ_IDX_PAGE_HEAD_SIZE;
} else {
bufPtr = POINTER_SHIFT(&idxBuf, idxBuf.head.writeOffset);
}
while(pHead != pNode) {
int nBytes = 0;
@ -308,18 +352,23 @@ int32_t tqStorePersist(TqMetaStore* pMeta) {
ASSERT(nBytesTxn == pSHead->ssize);
nBytes += nBytesTxn;
}
pNode->handle.offset = offset;
offset += nBytes;
//write idx file
//TODO: endian check and convert
*(bufPtr++) = pNode->handle.key;
*(bufPtr++) = pNode->handle.offset;
*(bufPtr++) = (int64_t)nBytes;
if((char*)(bufPtr + 3) > writeBuf + TQ_PAGE_SIZE) {
nBytes = write(pMeta->idxFd, writeBuf, sizeof(writeBuf));
idxBuf.head.writeOffset += TQ_IDX_SIZE;
if(idxBuf.head.writeOffset >= TQ_PAGE_SIZE) {
nBytes = write(pMeta->idxFd, &idxBuf, TQ_PAGE_SIZE);
//TODO: handle error with tfile
ASSERT(nBytes == sizeof(writeBuf));
memset(writeBuf, 0, TQ_PAGE_SIZE);
bufPtr = (int64_t*)writeBuf;
ASSERT(nBytes == TQ_PAGE_SIZE);
memset(&idxBuf, 0, TQ_PAGE_SIZE);
idxBuf.head.writeOffset = TQ_IDX_PAGE_HEAD_SIZE;
bufPtr = (int64_t*)&idxBuf.buffer;
}
//remove from unpersist list
pHead->unpersistNext = pNode->unpersistNext;
@ -331,7 +380,7 @@ int32_t tqStorePersist(TqMetaStore* pMeta) {
if(pNode->handle.valueInUse == TQ_DELETE_TOKEN &&
pNode->handle.valueInTxn == NULL
) {
int bucketKey = pNode->handle.key & TQ_BUCKET_SIZE;
int bucketKey = pNode->handle.key & TQ_BUCKET_MASK;
TqMetaList* pBucketHead = pMeta->bucket[bucketKey];
if(pBucketHead == pNode) {
pMeta->bucket[bucketKey] = pNode->next;
@ -351,11 +400,11 @@ int32_t tqStorePersist(TqMetaStore* pMeta) {
//write left bytes
free(pSHead);
if((char*)bufPtr != writeBuf) {
int used = (char*)bufPtr - writeBuf;
int nBytes = write(pMeta->idxFd, writeBuf, used);
//TODO: write new version in tfile
if((char*)bufPtr != idxBuf.buffer) {
int nBytes = write(pMeta->idxFd, &idxBuf, idxBuf.head.writeOffset);
//TODO: handle error in tfile
ASSERT(nBytes == used);
ASSERT(nBytes == idxBuf.head.writeOffset);
}
//TODO: using fsync in tfile
fsync(pMeta->idxFd);
@ -364,7 +413,7 @@ int32_t tqStorePersist(TqMetaStore* pMeta) {
}
static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
@ -397,11 +446,12 @@ static int32_t tqHandlePutCommitted(TqMetaStore* pMeta, int64_t key, void* value
}
void* tqHandleGet(TqMetaStore* pMeta, int64_t key) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
if(pNode->handle.valueInUse != NULL) {
if(pNode->handle.valueInUse != NULL
&& pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
return pNode->handle.valueInUse;
} else {
return NULL;
@ -414,7 +464,7 @@ void* tqHandleGet(TqMetaStore* pMeta, int64_t key) {
}
int32_t tqHandleMovePut(TqMetaStore* pMeta, int64_t key, void* value) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
@ -452,7 +502,7 @@ int32_t tqHandleCopyPut(TqMetaStore* pMeta, int64_t key, void* value, size_t vsi
return -1;
}
memcpy(vmem, value, vsize);
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
@ -484,7 +534,7 @@ int32_t tqHandleCopyPut(TqMetaStore* pMeta, int64_t key, void* value, size_t vsi
}
static void* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
@ -502,10 +552,13 @@ static void* tqHandleGetUncommitted(TqMetaStore* pMeta, int64_t key) {
}
int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
if(pNode->handle.valueInTxn == NULL) {
return -1;
}
if(pNode->handle.valueInUse
&& pNode->handle.valueInUse != TQ_DELETE_TOKEN) {
pMeta->deleter(pNode->handle.valueInUse);
@ -518,11 +571,11 @@ int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) {
pNode = pNode->next;
}
}
return -1;
return -2;
}
int32_t tqHandleAbort(TqMetaStore* pMeta, int64_t key) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.key == key) {
@ -543,12 +596,13 @@ int32_t tqHandleAbort(TqMetaStore* pMeta, int64_t key) {
}
int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) {
int64_t bucketKey = key & TQ_BUCKET_SIZE;
int64_t bucketKey = key & TQ_BUCKET_MASK;
TqMetaList* pNode = pMeta->bucket[bucketKey];
while(pNode) {
if(pNode->handle.valueInTxn
&& pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
pMeta->deleter(pNode->handle.valueInTxn);
if(pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
if(pNode->handle.valueInTxn) {
pMeta->deleter(pNode->handle.valueInTxn);
}
pNode->handle.valueInTxn = TQ_DELETE_TOKEN;
tqLinkUnpersist(pMeta, pNode);
return 0;
@ -560,34 +614,7 @@ int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) {
return -1;
}
int32_t tqHandleClear(TqMetaStore* pMeta, int64_t key) {
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;
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;
if(pNode->handle.valueInTxn != TQ_DELETE_TOKEN) {
pMeta->deleter(pNode->handle.valueInTxn);
}
pNode->handle.valueInTxn = TQ_DELETE_TOKEN;
}
if(exist) {
tqLinkUnpersist(pMeta, pNode);
return 0;
}
return -1;
} else {
pNode = pNode->next;
}
}
return -2;
//TODO: clean deleted idx and data from persistent file
int32_t tqStoreCompact(TqMetaStore *pMeta) {
return 0;
}

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

@ -57,6 +57,10 @@ TEST_F(TqMetaTest, copyPutTest) {
Foo* pFoo = (Foo*) tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true);
tqHandleCommit(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo->a, 3);
}
TEST_F(TqMetaTest, persistTest) {
@ -82,8 +86,6 @@ TEST_F(TqMetaTest, persistTest) {
pBar = (Foo*)tqHandleGet(pMeta, 2);
EXPECT_EQ(pBar == NULL, true);
//taosRemoveDir(pathName);
}
TEST_F(TqMetaTest, uncommittedTest) {
@ -130,4 +132,163 @@ TEST_F(TqMetaTest, deleteTest) {
tqHandleCommit(pMeta, 1);
pFoo = (Foo*) tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true);
tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter);
ASSERT(pMeta);
pFoo = (Foo*) tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo == NULL, true);
}
TEST_F(TqMetaTest, intxnPersist) {
Foo* pFoo = (Foo*)malloc(sizeof(Foo));
pFoo->a = 3;
tqHandleMovePut(pMeta, 1, pFoo);
tqHandleCommit(pMeta, 1);
Foo* pBar = (Foo*)malloc(sizeof(Foo));
pBar->a = 4;
tqHandleMovePut(pMeta, 1, pBar);
Foo* pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo1->a, 3);
tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter);
ASSERT(pMeta);
pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo1->a, 3);
tqHandleCommit(pMeta, 1);
pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo1->a, 4);
tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter);
ASSERT(pMeta);
pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
EXPECT_EQ(pFoo1->a, 4);
}
TEST_F(TqMetaTest, multiplePage) {
srand(0);
std::vector<int> v;
for(int i = 0; i < 1000; i++) {
v.push_back(rand());
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
}
for(int i = 0; i < 500; i++) {
tqHandleCommit(pMeta, i);
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]);
}
tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter);
ASSERT(pMeta);
for(int i = 500; i < 1000; i++) {
tqHandleCommit(pMeta, i);
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]);
}
for(int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]);
}
}
TEST_F(TqMetaTest, multipleRewrite) {
srand(0);
std::vector<int> v;
for(int i = 0; i < 1000; i++) {
v.push_back(rand());
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
}
for(int i = 0; i < 500; i++) {
tqHandleCommit(pMeta, i);
v[i] = rand();
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
}
for(int i = 500; i < 1000; i++) {
v[i] = rand();
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
}
for(int i = 0; i < 1000; i++) {
tqHandleCommit(pMeta, i);
}
tqStoreClose(pMeta);
pMeta = tqStoreOpen(pathName,
FooSerializer, FooDeserializer, FooDeleter);
ASSERT(pMeta);
for(int i = 500; i < 1000; i++) {
v[i] = rand();
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
tqHandleCommit(pMeta, i);
}
for(int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]);
}
}
TEST_F(TqMetaTest, dupCommit) {
srand(0);
std::vector<int> v;
for(int i = 0; i < 1000; i++) {
v.push_back(rand());
Foo foo;
foo.a = v[i];
tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
}
for(int i = 0; i < 1000; i++) {
int ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, 0);
ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, -1);
}
for(int i = 0; i < 1000; i++) {
int ret = tqHandleCommit(pMeta, i);
EXPECT_EQ(ret, -1);
}
for(int i = 0; i < 1000; i++) {
Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
EXPECT_EQ(pFoo->a, v[i]);
}
}

2
source/libs/sync/inc/raft.h
View File

@ -68,7 +68,7 @@ struct SSyncRaft {
int maxMsgSize;
SSyncRaftProgressTracker *tracker;
ESyncRole state;
ESyncState state;
// isLearner is true if the local raft node is a learner.
bool isLearner;

2
source/libs/sync/src/raft_handle_election_message.c
View File

@ -19,7 +19,7 @@
#include "raft_message.h"
int syncRaftHandleElectionMessage(SSyncRaft* pRaft, const SSyncMessage* pMsg) {
if (pRaft->state == TAOS_SYNC_ROLE_LEADER) {
if (pRaft->state == TAOS_SYNC_STATE_LEADER) {
syncDebug("[%d:%d] ignoring RAFT_MSG_INTERNAL_ELECTION because already leader", pRaft->selfGroupId, pRaft->selfId);
return 0;
}

4
source/libs/sync/src/raft_handle_vote_resp_message.c
View File

@ -23,7 +23,7 @@ int syncRaftHandleVoteRespMessage(SSyncRaft* pRaft, const SSyncMessage* pMsg) {
int quorum;
int voterIndex;
assert(pRaft->state == TAOS_SYNC_ROLE_CANDIDATE);
assert(pRaft->state == TAOS_SYNC_STATE_CANDIDATE);
voterIndex = syncRaftConfigurationIndexOfNode(pRaft, pMsg->from);
if (voterIndex == -1) {
@ -31,7 +31,7 @@ int syncRaftHandleVoteRespMessage(SSyncRaft* pRaft, const SSyncMessage* pMsg) {
return 0;
}
if (pRaft->state != TAOS_SYNC_ROLE_CANDIDATE) {
if (pRaft->state != TAOS_SYNC_STATE_CANDIDATE) {
syncError("[%d:%d] is not candidate, ignore vote resp", pRaft->selfGroupId, pRaft->selfId);
return 0;
}

2
source/libs/sync/src/raft_replication.c
View File

@ -23,7 +23,7 @@ static int sendAppendEntries(SSyncRaft* pRaft, int i, SyncIndex index, SyncTerm
int syncRaftReplicate(SSyncRaft* pRaft, int i) {
#if 0
assert(pRaft->state == TAOS_SYNC_ROLE_LEADER);
assert(pRaft->state == TAOS_SYNC_STATE_LEADER);
assert(i >= 0 && i < pRaft->leaderState.nProgress);
SyncNodeId nodeId = pRaft->cluster.nodeInfo[i].nodeId;

8
source/libs/sync/src/sync.c
View File

@ -167,6 +167,14 @@ int32_t syncPropose(SSyncNode* syncNode, const SSyncBuffer* pBuf, void* pData, b
void syncReconfig(const SSyncNode* pNode, const SSyncCluster* pCfg) {}
int32_t syncAddNode(SSyncNode syncNode, const SNodeInfo *pNode) {
return 0;
}
int32_t syncRemoveNode(SSyncNode syncNode, const SNodeInfo *pNode) {
return 0;
}
// process rpc rsp message from other sync server
static void syncProcessRsp(SRpcMsg *pMsg, SEpSet *pEpSet) {

12
source/libs/sync/src/sync_raft_impl.c
View File

@ -44,7 +44,7 @@ void syncRaftBecomeFollower(SSyncRaft* pRaft, SyncTerm term, SyncNodeId leaderId
resetRaft(pRaft, term);
pRaft->tickFp = tickElection;
pRaft->leaderId = leaderId;
pRaft->state = TAOS_SYNC_ROLE_FOLLOWER;
pRaft->state = TAOS_SYNC_STATE_FOLLOWER;
syncInfo("[%d:%d] became followe at term %" PRId64 "", pRaft->selfGroupId, pRaft->selfId, pRaft->term);
}
@ -58,7 +58,7 @@ void syncRaftBecomePreCandidate(SSyncRaft* pRaft) {
**/
pRaft->stepFp = stepCandidate;
pRaft->tickFp = tickElection;
pRaft->state = TAOS_SYNC_ROLE_CANDIDATE;
pRaft->state = TAOS_SYNC_STATE_CANDIDATE;
pRaft->candidateState.inPreVote = true;
syncInfo("[%d:%d] became pre-candidate at term %" PRId64 "", pRaft->selfGroupId, pRaft->selfId, pRaft->term);
}
@ -72,17 +72,17 @@ void syncRaftBecomeCandidate(SSyncRaft* pRaft) {
resetRaft(pRaft, pRaft->term + 1);
pRaft->tickFp = tickElection;
pRaft->voteFor = pRaft->selfId;
pRaft->state = TAOS_SYNC_ROLE_CANDIDATE;
pRaft->state = TAOS_SYNC_STATE_CANDIDATE;
syncInfo("[%d:%d] became candidate at term %" PRId64 "", pRaft->selfGroupId, pRaft->selfId, pRaft->term);
}
void syncRaftBecomeLeader(SSyncRaft* pRaft) {
assert(pRaft->state != TAOS_SYNC_ROLE_FOLLOWER);
assert(pRaft->state != TAOS_SYNC_STATE_FOLLOWER);
pRaft->stepFp = stepLeader;
resetRaft(pRaft, pRaft->term);
pRaft->leaderId = pRaft->leaderId;
pRaft->state = TAOS_SYNC_ROLE_LEADER;
pRaft->state = TAOS_SYNC_STATE_LEADER;
// TODO: check if there is pending config log
int nPendingConf = syncRaftLogNumOfPendingConf(pRaft->log);
if (nPendingConf > 1) {
@ -263,7 +263,7 @@ static bool maybeCommit(SSyncRaft* pRaft) {
* trigger I/O requests for newly appended log entries or heartbeats.
**/
static int triggerAll(SSyncRaft* pRaft) {
assert(pRaft->state == TAOS_SYNC_ROLE_LEADER);
assert(pRaft->state == TAOS_SYNC_STATE_LEADER);
int i;
for (i = 0; i < pRaft->cluster.replica; ++i) {