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

enh: adjust sync propose

This commit is contained in:
Shengliang Guan 2022-11-07 20:31:26 +08:00
parent 46e55ba9fb
commit 4ddd25a29c
11 changed files with 29 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/libs/sync/inc/syncTools.h
View File

@ -182,9 +182,8 @@ typedef struct SyncClientRequest {
char data[]; // origin RpcMsg.pCont
} SyncClientRequest;
SyncClientRequest* syncClientRequestBuild(uint32_t dataLen);
SyncClientRequest* syncClientRequestBuild2(const SRpcMsg* pOriginalRpcMsg, uint64_t seqNum, bool isWeak,
int32_t vgId); // step 1
SyncClientRequest* syncClientRequestAlloc(uint32_t dataLen);
SyncClientRequest* syncClientRequestBuild(const SRpcMsg* pMsg, uint64_t seqNum, bool isWeak, int32_t vgId); // step 1
void syncClientRequestDestroy(SyncClientRequest* pMsg);
void syncClientRequestSerialize(const SyncClientRequest* pMsg, char* buf, uint32_t bufLen);
void syncClientRequestDeserialize(const char* buf, uint32_t len, SyncClientRequest* pMsg);

19
source/libs/sync/src/syncMain.c
View File

@ -669,13 +669,11 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
goto _END;
}
SRespStub stub;
stub.createTime = taosGetTimestampMs();
stub.rpcMsg = *pMsg;
uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
SRespStub stub = {.createTime = taosGetTimestampMs(), .rpcMsg = *pMsg};
uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, seqNum, isWeak, pSyncNode->vgId);
SRpcMsg rpcMsg;
SyncClientRequest* pSyncMsg = syncClientRequestBuild(pMsg, seqNum, isWeak, pSyncNode->vgId);
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
// optimized one replica
@ -696,12 +694,9 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
sError("vgId:%d, failed to sync optimize index:%" PRId64 ", type:%s", pSyncNode->vgId, retIndex,
TMSG_INFO(pMsg->msgType));
}
} else {
if (pSyncNode->syncEqMsg != NULL && (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
ret = 0;
} else {
ret = -1;
ret = (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg);
if (ret != 0) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
sError("vgId:%d, failed to enqueue msg since its null", pSyncNode->vgId);
}
@ -2322,7 +2317,7 @@ static int32_t syncNodeEqNoop(SSyncNode* ths) {
uint32_t entryLen;
char* serialized = syncEntrySerialize(pEntry, &entryLen);
SyncClientRequest* pSyncMsg = syncClientRequestBuild(entryLen);
SyncClientRequest* pSyncMsg = syncClientRequestAlloc(entryLen);
ASSERT(pSyncMsg->dataLen == entryLen);
memcpy(pSyncMsg->data, serialized, entryLen);

10
source/libs/sync/src/syncMessage.c
View File

@ -831,10 +831,9 @@ void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) {
}
// ---- message process SyncClientRequest----
SyncClientRequest* syncClientRequestBuild(uint32_t dataLen) {
SyncClientRequest* syncClientRequestAlloc(uint32_t dataLen) {
uint32_t bytes = sizeof(SyncClientRequest) + dataLen;
SyncClientRequest* pMsg = taosMemoryMalloc(bytes);
memset(pMsg, 0, bytes);
SyncClientRequest* pMsg = taosMemoryCalloc(1, bytes);
pMsg->bytes = bytes;
pMsg->msgType = TDMT_SYNC_CLIENT_REQUEST;
pMsg->seqNum = 0;
@ -844,8 +843,8 @@ SyncClientRequest* syncClientRequestBuild(uint32_t dataLen) {
}
// step 1. original SRpcMsg => SyncClientRequest, add seqNum, isWeak
SyncClientRequest* syncClientRequestBuild2(const SRpcMsg* pOriginalRpcMsg, uint64_t seqNum, bool isWeak, int32_t vgId) {
SyncClientRequest* pMsg = syncClientRequestBuild(pOriginalRpcMsg->contLen);
SyncClientRequest* syncClientRequestBuild(const SRpcMsg* pOriginalRpcMsg, uint64_t seqNum, bool isWeak, int32_t vgId) {
SyncClientRequest* pMsg = syncClientRequestAlloc(pOriginalRpcMsg->contLen);
pMsg->vgId = vgId;
pMsg->originalRpcType = pOriginalRpcMsg->msgType;
pMsg->seqNum = seqNum;
@ -891,7 +890,6 @@ SyncClientRequest* syncClientRequestDeserialize2(const char* buf, uint32_t len)
// step 2. SyncClientRequest => RpcMsg, to queue
void syncClientRequest2RpcMsg(const SyncClientRequest* pMsg, SRpcMsg* pRpcMsg) {
memset(pRpcMsg, 0, sizeof(*pRpcMsg));
pRpcMsg->msgType = pMsg->msgType;
pRpcMsg->contLen = pMsg->bytes;
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);

2
source/libs/sync/test/syncApplyMsgTest.cpp
View File

@ -81,7 +81,7 @@ void test4() {
void test5() {
SyncApplyMsg *pMsg = createMsg();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncApplyMsg2RpcMsg(pMsg, &rpcMsg);
SyncApplyMsg *pMsg2 = syncApplyMsgFromRpcMsg2(&rpcMsg);
syncApplyMsgLog2((char *)"test5: syncClientRequest2RpcMsg -> syncApplyMsgFromRpcMsg2 ", pMsg2);

2
source/libs/sync/test/syncClientRequestBatchTest.cpp
View File

@ -59,7 +59,7 @@ void test2() {
uint32_t len = pMsg->bytes;
char * serialized = (char *)taosMemoryMalloc(len);
syncClientRequestSerialize(pMsg, serialized, len);
SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg->dataLen);
SyncClientRequest *pMsg2 = syncClientRequestAlloc(pMsg->dataLen);
syncClientRequestDeserialize(serialized, len, pMsg2);
syncClientRequestLog2((char *)"test2: syncClientRequestSerialize -> syncClientRequestDeserialize ", pMsg2);

8
source/libs/sync/test/syncClientRequestTest.cpp
View File

@ -21,7 +21,7 @@ SyncClientRequest *createMsg() {
rpcMsg.contLen = 20;
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
strcpy((char *)rpcMsg.pCont, "hello rpc");
SyncClientRequest *pMsg = syncClientRequestBuild2(&rpcMsg, 123, true, 1000);
SyncClientRequest *pMsg = syncClientRequestBuild(&rpcMsg, 123, true, 1000);
rpcFreeCont(rpcMsg.pCont);
return pMsg;
}
@ -37,7 +37,7 @@ void test2() {
uint32_t len = pMsg->bytes;
char *serialized = (char *)taosMemoryMalloc(len);
syncClientRequestSerialize(pMsg, serialized, len);
SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg->dataLen);
SyncClientRequest *pMsg2 = syncClientRequestAlloc(pMsg->dataLen);
syncClientRequestDeserialize(serialized, len, pMsg2);
syncClientRequestLog2((char *)"test2: syncClientRequestSerialize -> syncClientRequestDeserialize ", pMsg2);
@ -60,7 +60,7 @@ void test3() {
void test4() {
SyncClientRequest *pMsg = createMsg();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pMsg, &rpcMsg);
SyncClientRequest *pMsg2 = (SyncClientRequest *)taosMemoryMalloc(rpcMsg.contLen);
syncClientRequestFromRpcMsg(&rpcMsg, pMsg2);
@ -73,7 +73,7 @@ void test4() {
void test5() {
SyncClientRequest *pMsg = createMsg();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pMsg, &rpcMsg);
SyncClientRequest *pMsg2 = syncClientRequestFromRpcMsg2(&rpcMsg);
syncClientRequestLog2((char *)"test5: syncClientRequest2RpcMsg -> syncClientRequestFromRpcMsg2 ", pMsg2);

4
source/libs/sync/test/syncEncodeTest.cpp
View File

@ -102,12 +102,12 @@ SRpcMsg *step0() {
}
SyncClientRequest *step1(const SRpcMsg *pMsg) {
SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true, 1000);
SyncClientRequest *pRetMsg = syncClientRequestBuild(pMsg, 123, true, 1000);
return pRetMsg;
}
SRpcMsg *step2(const SyncClientRequest *pMsg) {
SRpcMsg *pRetMsg = (SRpcMsg *)taosMemoryMalloc(sizeof(SRpcMsg));
SRpcMsg *pRetMsg = (SRpcMsg *)taosMemoryCalloc(sizeof(SRpcMsg), 1);
syncClientRequest2RpcMsg(pMsg, pRetMsg);
return pRetMsg;
}

4
source/libs/sync/test/syncEntryTest.cpp
View File

@ -32,7 +32,7 @@ void test1() {
}
void test2() {
SyncClientRequest* pSyncMsg = syncClientRequestBuild(10);
SyncClientRequest* pSyncMsg = syncClientRequestAlloc(10);
pSyncMsg->originalRpcType = 33;
pSyncMsg->seqNum = 11;
pSyncMsg->isWeak = 1;
@ -46,7 +46,7 @@ void test2() {
}
void test3() {
SyncClientRequest* pSyncMsg = syncClientRequestBuild(10);
SyncClientRequest* pSyncMsg = syncClientRequestAlloc(10);
pSyncMsg->originalRpcType = 33;
pSyncMsg->seqNum = 11;
pSyncMsg->isWeak = 1;

4
source/libs/sync/test/syncRpcMsgTest.cpp
View File

@ -47,7 +47,7 @@ SyncClientRequest *createSyncClientRequest() {
rpcMsg.contLen = 20;
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
strcpy((char *)rpcMsg.pCont, "hello rpc");
SyncClientRequest *pMsg = syncClientRequestBuild2(&rpcMsg, 123, true, 1000);
SyncClientRequest *pMsg = syncClientRequestBuild(&rpcMsg, 123, true, 1000);
return pMsg;
}
@ -156,7 +156,7 @@ void test7() {
void test8() {
SyncClientRequest *pMsg = createSyncClientRequest();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pMsg, &rpcMsg);
syncRpcMsgLog2((char *)"test8", &rpcMsg);
syncClientRequestDestroy(pMsg);

4
source/libs/sync/test/syncSnapshotTest.cpp
View File

@ -162,7 +162,7 @@ SRpcMsg *step0() {
}
SyncClientRequest *step1(const SRpcMsg *pMsg) {
SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true, 1000);
SyncClientRequest *pRetMsg = syncClientRequestBuild(pMsg, 123, true, 1000);
return pRetMsg;
}
@ -206,7 +206,7 @@ int main(int argc, char **argv) {
for (int i = 0; i < 10; ++i) {
SyncClientRequest *pSyncClientRequest = pMsg1;
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pSyncClientRequest, &rpcMsg);
gSyncNode->syncEqMsg(gSyncNode->msgcb, &rpcMsg);

4
source/libs/sync/test/syncWriteTest.cpp
View File

@ -140,7 +140,7 @@ SRpcMsg *step0() {
}
SyncClientRequest *step1(const SRpcMsg *pMsg) {
SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true, 1000);
SyncClientRequest *pRetMsg = syncClientRequestBuild(pMsg, 123, true, 1000);
return pRetMsg;
}
@ -181,7 +181,7 @@ int main(int argc, char **argv) {
for (int i = 0; i < 10; ++i) {
SyncClientRequest *pSyncClientRequest = pMsg1;
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
syncClientRequest2RpcMsg(pSyncClientRequest, &rpcMsg);
gSyncNode->syncEqMsg(gSyncNode->msgcb, &rpcMsg);