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

refact: adjust sync log

This commit is contained in:
Shengliang Guan 2022-11-07 13:51:37 +08:00
parent 76649005dd
commit a1c8db0f88
25 changed files with 568 additions and 1243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
source/libs/sync/inc/syncInt.h
View File

@ -40,6 +40,28 @@ extern "C" {
#define sInfoLong(...) do { if (sDebugFlag & DEBUG_INFO) { taosPrintLongString("SYN ", DEBUG_INFO, 255, __VA_ARGS__); }} while(0)
#define sDebugLong(...) do { if (sDebugFlag & DEBUG_DEBUG) { taosPrintLongString("SYN ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); }} while(0)
#define sTraceLong(...) do { if (sDebugFlag & DEBUG_TRACE) { taosPrintLongString("SYN ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); }} while(0)
#define sNFatal(pNode, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintNodeLog("SYN FATAL ", DEBUG_FATAL, 255, pNode, __VA_ARGS__); }
#define sNError(pNode, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintNodeLog("SYN ERROR ", DEBUG_ERROR, 255, pNode, __VA_ARGS__); }
#define sNWarn(pNode, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintNodeLog("SYN WARN ", DEBUG_WARN, 255, pNode, __VA_ARGS__); }
#define sNInfo(pNode, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintNodeLog("SYN ", DEBUG_INFO, 255, pNode, __VA_ARGS__); }
#define sNDebug(pNode, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintNodeLog("SYN ", DEBUG_DEBUG, sDebugFlag, pNode, __VA_ARGS__); }
#define sNTrace(pNode, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintNodeLog("SYN ", DEBUG_TRACE, sDebugFlag, pNode, __VA_ARGS__); }
#define sSFatal(pSender, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotSenderLog("SYN FATAL ", DEBUG_FATAL, 255, pSender, __VA_ARGS__); }
#define sSError(pSender, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotSenderLog("SYN ERROR ", DEBUG_ERROR, 255, pSender, __VA_ARGS__); }
#define sSWarn(pSender, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotSenderLog("SYN WARN ", DEBUG_WARN, 255, pSender, __VA_ARGS__); }
#define sSInfo(pSender, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotSenderLog("SYN ", DEBUG_INFO, 255, pSender, __VA_ARGS__); }
#define sSDebug(pSender, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotSenderLog("SYN ", DEBUG_DEBUG, sDebugFlag, pSender, __VA_ARGS__); }
#define sSTrace(pSender, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotSenderLog("SYN ", DEBUG_TRACE, sDebugFlag, pSender, __VA_ARGS__); }
#define sRFatal(pReceiver, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotReceiverLog("SYN FATAL ", DEBUG_FATAL, 255, pReceiver, __VA_ARGS__); }
#define sRError(pReceiver, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotReceiverLog("SYN ERROR ", DEBUG_ERROR, 255, pReceiver, __VA_ARGS__); }
#define sRWarn(pReceiver, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotReceiverLog("SYN WARN ", DEBUG_WARN, 255, pReceiver, __VA_ARGS__); }
#define sRInfo(pReceiver, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_INFO, 255, pReceiver, __VA_ARGS__); }
#define sRDebug(pReceiver, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_DEBUG, sDebugFlag, pReceiver, __VA_ARGS__); }
#define sRTrace(pReceiver, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_TRACE, sDebugFlag, pReceiver, __VA_ARGS__); }
// clang-format on
typedef struct SyncTimeout SyncTimeout;
@ -250,8 +272,6 @@ int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, S
int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
cJSON* syncNode2Json(const SSyncNode* pSyncNode);
char* syncNode2Str(const SSyncNode* pSyncNode);
void syncNodeEventLog(const SSyncNode* pSyncNode, char* str);
void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str);
char* syncNode2SimpleStr(const SSyncNode* pSyncNode);
bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config);
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex);
@ -357,6 +377,13 @@ void syncNodeLog(SSyncNode* pObj);
void syncNodeLog2(char* s, SSyncNode* pObj);
void syncNodeLog3(char* s, SSyncNode* pObj);
// syncUtil.h
void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...);
void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender,
const char* format, ...);
void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver,
const char* format, ...);
#ifdef __cplusplus
}
#endif

30
source/libs/sync/inc/syncRaftCfg.h
View File

@ -45,8 +45,8 @@ int32_t raftCfgIndexClose(SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexPersist(SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexAddConfigIndex(SRaftCfgIndex *pRaftCfgIndex, SyncIndex configIndex);
cJSON * raftCfgIndex2Json(SRaftCfgIndex *pRaftCfgIndex);
char * raftCfgIndex2Str(SRaftCfgIndex *pRaftCfgIndex);
cJSON *raftCfgIndex2Json(SRaftCfgIndex *pRaftCfgIndex);
char *raftCfgIndex2Str(SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexFromJson(const cJSON *pRoot, SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexFromStr(const char *s, SRaftCfgIndex *pRaftCfgIndex);
@ -73,14 +73,14 @@ int32_t raftCfgClose(SRaftCfg *pRaftCfg);
int32_t raftCfgPersist(SRaftCfg *pRaftCfg);
int32_t raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex);
cJSON * syncCfg2Json(SSyncCfg *pSyncCfg);
char * syncCfg2Str(SSyncCfg *pSyncCfg);
char * syncCfg2SimpleStr(SSyncCfg *pSyncCfg);
cJSON *syncCfg2Json(SSyncCfg *pSyncCfg);
char *syncCfg2Str(SSyncCfg *pSyncCfg);
void syncCfg2SimpleStr(const SSyncCfg *pCfg, char *str, int32_t bufLen);
int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg);
int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg);
cJSON * raftCfg2Json(SRaftCfg *pRaftCfg);
char * raftCfg2Str(SRaftCfg *pRaftCfg);
cJSON *raftCfg2Json(SRaftCfg *pRaftCfg);
char *raftCfg2Str(SRaftCfg *pRaftCfg);
int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg);
int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg);
@ -93,21 +93,7 @@ typedef struct SRaftCfgMeta {
int32_t raftCfgCreateFile(SSyncCfg *pCfg, SRaftCfgMeta meta, const char *path);
// for debug ----------------------
void syncCfgPrint(SSyncCfg *pCfg);
void syncCfgPrint2(char *s, SSyncCfg *pCfg);
void syncCfgLog(SSyncCfg *pCfg);
void syncCfgLog2(char *s, SSyncCfg *pCfg);
void syncCfgLog3(char *s, SSyncCfg *pCfg);
void raftCfgPrint(SRaftCfg *pCfg);
void raftCfgPrint2(char *s, SRaftCfg *pCfg);
void raftCfgLog(SRaftCfg *pCfg);
void raftCfgLog2(char *s, SRaftCfg *pCfg);
void raftCfgIndexPrint(SRaftCfgIndex *pCfg);
void raftCfgIndexPrint2(char *s, SRaftCfgIndex *pCfg);
void raftCfgIndexLog(SRaftCfgIndex *pCfg);
// for debug -------------------
void raftCfgIndexLog2(char *s, SRaftCfgIndex *pCfg);
#ifdef __cplusplus

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

@ -67,7 +67,6 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender);
cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender);
char *snapshotSender2Str(SSyncSnapshotSender *pSender);
char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event);
int32_t syncNodeStartSnapshot(SSyncNode *pSyncNode, SRaftId *pDestId);
@ -96,7 +95,6 @@ void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceive
cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event);
//---------------------------------------------------
// on message

35
source/libs/sync/inc/syncUtil.h
View File

@ -20,48 +20,35 @@
extern "C" {
#endif
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include "syncInt.h"
#include "syncMessage.h"
#include "taosdef.h"
// ---- encode / decode
uint64_t syncUtilAddr2U64(const char* host, uint16_t port);
void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port);
void syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet);
void syncUtilU642Addr(uint64_t u64, char* host, int64_t len, uint16_t* port);
void syncUtilnodeInfo2EpSet(const SNodeInfo* pInfo, SEpSet* pEpSet);
void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet);
bool syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
bool syncUtilnodeInfo2raftId(const SNodeInfo* pInfo, SyncGroupId vgId, SRaftId* raftId);
bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
bool syncUtilEmptyId(const SRaftId* pId);
// ---- SSyncBuffer ----
void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len);
void syncUtilbufDestroy(SSyncBuffer* syncBuf);
void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest);
void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest);
// ---- misc ----
int32_t syncUtilRand(int32_t max);
int32_t syncUtilElectRandomMS(int32_t min, int32_t max);
int32_t syncUtilQuorum(int32_t replicaNum);
cJSON* syncUtilNodeInfo2Json(const SNodeInfo* p);
cJSON* syncUtilRaftId2Json(const SRaftId* p);
char* syncUtilRaftId2Str(const SRaftId* p);
const char* syncStr(ESyncState state);
bool syncUtilCanPrint(char c);
char* syncUtilprintBin(char* ptr, uint32_t len);
char* syncUtilprintBin2(char* ptr, uint32_t len);
SyncIndex syncUtilMinIndex(SyncIndex a, SyncIndex b);
SyncIndex syncUtilMaxIndex(SyncIndex a, SyncIndex b);
char* syncUtilPrintBin(char* ptr, uint32_t len);
char* syncUtilPrintBin2(char* ptr, uint32_t len);
void syncUtilMsgHtoN(void* msg);
void syncUtilMsgNtoH(void* msg);
bool syncUtilIsData(tmsg_t msgType);
bool syncUtilUserPreCommit(tmsg_t msgType);
bool syncUtilUserCommit(tmsg_t msgType);
bool syncUtilUserRollback(tmsg_t msgType);
void syncUtilJson2Line(char* jsonStr);
void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...);
void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender,
const char* format, ...);
void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver,
const char* format, ...);
#ifdef __cplusplus
}

14
source/libs/sync/src/syncAppendEntries.c
View File

@ -91,7 +91,7 @@
int32_t syncNodeFollowerCommit(SSyncNode* ths, SyncIndex newCommitIndex) {
if (ths->state != TAOS_SYNC_STATE_FOLLOWER) {
syncNodeEventLog(ths, "can not do follower commit");
sNTrace(ths, "can not do follower commit");
return -1;
}
@ -106,11 +106,7 @@ int32_t syncNodeFollowerCommit(SSyncNode* ths, SyncIndex newCommitIndex) {
SyncIndex commitBegin = ths->commitIndex;
SyncIndex commitEnd = snapshot.lastApplyIndex;
ths->commitIndex = snapshot.lastApplyIndex;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin,
commitEnd);
syncNodeEventLog(ths, eventLog);
sNTrace(ths, "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin, commitEnd);
}
SyncIndex beginIndex = ths->commitIndex + 1;
@ -192,11 +188,7 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, SyncAppendEntries* pMsg) {
if (code == 0) {
if (pLocalEntry->term == pAppendEntry->term) {
// do nothing
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "log match, do nothing, index:%" PRId64, appendIndex);
syncNodeEventLog(ths, logBuf);
sNTrace(ths, "log match, do nothing, index:%" PRId64, appendIndex);
} else {
// truncate
code = ths->pLogStore->syncLogTruncate(ths->pLogStore, appendIndex);

14
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -43,12 +43,7 @@
static void syncNodeStartSnapshotOnce(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm,
SyncAppendEntriesReply* pMsg) {
if (beginIndex > endIndex) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "snapshot param error, start:%" PRId64 ", end:%" PRId64, beginIndex, endIndex);
syncNodeErrorLog(ths, logBuf);
} while (0);
sNError(ths, "snapshot param error, start:%" PRId64 ", end:%" PRId64, beginIndex, endIndex);
return;
}
@ -57,12 +52,7 @@ static void syncNodeStartSnapshotOnce(SSyncNode* ths, SyncIndex beginIndex, Sync
ASSERT(pSender != NULL);
if (snapshotSenderIsStart(pSender)) {
do {
char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender already start");
syncNodeErrorLog(ths, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSError(pSender, "snapshot sender already start");
return;
}

44
source/libs/sync/src/syncCommit.c
View File

@ -51,12 +51,12 @@ void syncOneReplicaAdvance(SSyncNode* pSyncNode) {
}
if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
syncNodeErrorLog(pSyncNode, "not leader, can not advance commit index");
sNError(pSyncNode, "not leader, can not advance commit index");
return;
}
if (pSyncNode->replicaNum != 1) {
syncNodeErrorLog(pSyncNode, "not one replica, can not advance commit index");
sNError(pSyncNode, "not one replica, can not advance commit index");
return;
}
@ -67,23 +67,13 @@ void syncOneReplicaAdvance(SSyncNode* pSyncNode) {
SyncIndex commitBegin = pSyncNode->commitIndex;
SyncIndex commitEnd = snapshot.lastApplyIndex;
pSyncNode->commitIndex = snapshot.lastApplyIndex;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin,
commitEnd);
syncNodeEventLog(pSyncNode, eventLog);
sNTrace(pSyncNode, "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin, commitEnd);
}
// advance commit index as large as possible
SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode);
if (lastIndex > pSyncNode->commitIndex) {
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "commit by wal from index:%" PRId64 " to index:%" PRId64,
pSyncNode->commitIndex + 1, lastIndex);
syncNodeEventLog(pSyncNode, eventLog);
} while (0);
sNTrace(pSyncNode, "commit by wal from index:%" PRId64 " to index:%" PRId64, pSyncNode->commitIndex + 1, lastIndex);
pSyncNode->commitIndex = lastIndex;
}
@ -101,7 +91,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
}
if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
syncNodeErrorLog(pSyncNode, "not leader, can not advance commit index");
sNError(pSyncNode, "not leader, can not advance commit index");
return;
}
@ -112,11 +102,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
SyncIndex commitBegin = pSyncNode->commitIndex;
SyncIndex commitEnd = snapshot.lastApplyIndex;
pSyncNode->commitIndex = snapshot.lastApplyIndex;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin,
commitEnd);
syncNodeEventLog(pSyncNode, eventLog);
sNTrace(pSyncNode, "commit by snapshot from index:%" PRId64 " to index:%" PRId64, commitBegin, commitEnd);
}
// update commit index
@ -134,9 +120,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
} else {
int32_t code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index, &pEntry);
if (code != 0) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "advance commit index error, read wal index:%" PRId64, index);
syncNodeErrorLog(pSyncNode, logBuf);
sNError(pSyncNode, "advance commit index error, read wal index:%" PRId64, index);
return;
}
}
@ -153,12 +137,8 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
break;
} else {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "can not commit due to term not equal, index:%" PRId64 ", term:%" PRIu64,
pEntry->index, pEntry->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode, "can not commit due to term not equal, index:%" PRId64 ", term:%" PRIu64, pEntry->index,
pEntry->term);
}
if (h) {
@ -190,10 +170,8 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
if (pSyncNode->pFsm != NULL) {
int32_t code = syncNodeDoCommit(pSyncNode, beginIndex, endIndex, pSyncNode->state);
if (code != 0) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "advance commit index error, do commit begin:%" PRId64 ", end:%" PRId64,
beginIndex, endIndex);
syncNodeErrorLog(pSyncNode, logBuf);
sNError(pSyncNode, "advance commit index error, do commit begin:%" PRId64 ", end:%" PRId64, beginIndex,
endIndex);
return;
}
}

6
source/libs/sync/src/syncElection.c
View File

@ -32,7 +32,7 @@
// /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
int32_t syncNodeElect(SSyncNode* pSyncNode) {
syncNodeEventLog(pSyncNode, "begin election");
sNTrace(pSyncNode, "begin election");
int32_t ret = 0;
if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
@ -40,7 +40,7 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
}
if (pSyncNode->state != TAOS_SYNC_STATE_CANDIDATE) {
syncNodeErrorLog(pSyncNode, "not candidate, can not elect");
sNError(pSyncNode, "not candidate, can not elect");
return -1;
}
@ -82,7 +82,7 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode) {
if (pSyncNode->state != TAOS_SYNC_STATE_CANDIDATE) {
syncNodeEventLog(pSyncNode, "not candidate, stop elect");
sNTrace(pSyncNode, "not candidate, stop elect");
return 0;
}

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

File diff suppressed because it is too large Load Diff

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

@ -85,20 +85,20 @@ cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
} else if (pRpcMsg->msgType == TDMT_SYNC_COMMON_RESPONSE) {
pRoot = cJSON_CreateObject();
char* s;
s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
s = syncUtilPrintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
cJSON_AddStringToObject(pRoot, "pCont", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
s = syncUtilPrintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
cJSON_AddStringToObject(pRoot, "pCont2", s);
taosMemoryFree(s);
} else {
pRoot = cJSON_CreateObject();
char* s;
s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
s = syncUtilPrintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
cJSON_AddStringToObject(pRoot, "pCont", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
s = syncUtilPrintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
cJSON_AddStringToObject(pRoot, "pCont2", s);
taosMemoryFree(s);
}
@ -511,10 +511,10 @@ cJSON* syncPing2Json(const SyncPing* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -781,10 +781,10 @@ cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -924,10 +924,10 @@ cJSON* syncClientRequest2Json(const SyncClientRequest* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -1094,10 +1094,10 @@ cJSON* syncClientRequestBatch2Json(const SyncClientRequestBatch* pMsg) {
}
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -1564,10 +1564,10 @@ cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -1803,10 +1803,10 @@ cJSON* syncAppendEntriesBatch2Json(const SyncAppendEntriesBatch* pMsg) {
}
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -2716,10 +2716,10 @@ cJSON* syncApplyMsg2Json(const SyncApplyMsg* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -2893,10 +2893,10 @@ cJSON* syncSnapshotSend2Json(const SyncSnapshotSend* pMsg) {
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
s = syncUtilPrintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}

114
source/libs/sync/src/syncRaftCfg.c
View File

@ -249,31 +249,16 @@ char *syncCfg2Str(SSyncCfg *pSyncCfg) {
return serialized;
}
char *syncCfg2SimpleStr(SSyncCfg *pSyncCfg) {
if (pSyncCfg != NULL) {
int32_t len = 512;
char *s = taosMemoryMalloc(len);
memset(s, 0, len);
void syncCfg2SimpleStr(const SSyncCfg *pCfg, char *buf, int32_t bufLen) {
int32_t len = snprintf(buf, bufLen, "{r-num:%d, my:%d, ", pCfg->replicaNum, pCfg->myIndex);
snprintf(s, len, "{r-num:%d, my:%d, ", pSyncCfg->replicaNum, pSyncCfg->myIndex);
char *p = s + strlen(s);
for (int i = 0; i < pSyncCfg->replicaNum; ++i) {
/*
if (p + 128 + 32 > s + len) {
break;
}
*/
char buf[128 + 32];
snprintf(buf, sizeof(buf), "%s:%d, ", pSyncCfg->nodeInfo[i].nodeFqdn, pSyncCfg->nodeInfo[i].nodePort);
strncpy(p, buf, sizeof(buf));
p = s + strlen(s);
for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
if (i < pCfg->replicaNum - 1) {
len += snprintf(buf + len, bufLen - len, "%s:%d, ", pCfg->nodeInfo[i].nodeFqdn, pCfg->nodeInfo[i].nodePort);
} else {
len += snprintf(buf + len, bufLen - len, "%s:%d}", pCfg->nodeInfo[i].nodeFqdn, pCfg->nodeInfo[i].nodePort);
}
strcpy(p - 2, "}");
return s;
}
return NULL;
}
int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg) {
@ -442,88 +427,3 @@ int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg) {
return 0;
}
// for debug ----------------------
void syncCfgPrint(SSyncCfg *pCfg) {
char *serialized = syncCfg2Str(pCfg);
printf("syncCfgPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void syncCfgPrint2(char *s, SSyncCfg *pCfg) {
char *serialized = syncCfg2Str(pCfg);
printf("syncCfgPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void syncCfgLog(SSyncCfg *pCfg) {
char *serialized = syncCfg2Str(pCfg);
sTrace("syncCfgLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void syncCfgLog2(char *s, SSyncCfg *pCfg) {
char *serialized = syncCfg2Str(pCfg);
sTrace("syncCfgLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
void syncCfgLog3(char *s, SSyncCfg *pCfg) {
char *serialized = syncCfg2SimpleStr(pCfg);
sTrace("syncCfgLog3 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
void raftCfgPrint(SRaftCfg *pCfg) {
char *serialized = raftCfg2Str(pCfg);
printf("raftCfgPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftCfgPrint2(char *s, SRaftCfg *pCfg) {
char *serialized = raftCfg2Str(pCfg);
printf("raftCfgPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftCfgLog(SRaftCfg *pCfg) {
char *serialized = raftCfg2Str(pCfg);
sTrace("raftCfgLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void raftCfgLog2(char *s, SRaftCfg *pCfg) {
char *serialized = raftCfg2Str(pCfg);
sTrace("raftCfgLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
// ---------
void raftCfgIndexPrint(SRaftCfgIndex *pCfg) {
char *serialized = raftCfgIndex2Str(pCfg);
printf("raftCfgIndexPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftCfgIndexPrint2(char *s, SRaftCfgIndex *pCfg) {
char *serialized = raftCfgIndex2Str(pCfg);
printf("raftCfgIndexPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftCfgIndexLog(SRaftCfgIndex *pCfg) {
char *serialized = raftCfgIndex2Str(pCfg);
sTrace("raftCfgIndexLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void raftCfgIndexLog2(char *s, SRaftCfgIndex *pCfg) {
char *serialized = raftCfgIndex2Str(pCfg);
sTrace("raftCfgIndexLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}

120
source/libs/sync/src/syncRaftEntry.c
View File

@ -144,11 +144,11 @@ cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen);
char* s;
s = syncUtilprintBin((char*)(pEntry->data), pEntry->dataLen);
s = syncUtilPrintBin((char*)(pEntry->data), pEntry->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char*)(pEntry->data), pEntry->dataLen);
s = syncUtilPrintBin2((char*)(pEntry->data), pEntry->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
taosMemoryFree(s);
}
@ -226,9 +226,9 @@ SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
void raftCacheDestroy(SRaftEntryHashCache* pCache) {
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
taosHashCleanup(pCache->pEntryHash);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
taosThreadMutexDestroy(&(pCache->mutex));
taosMemoryFree(pCache);
}
@ -238,25 +238,20 @@ void raftCacheDestroy(SRaftEntryHashCache* pCache) {
// max count, return 0
// error, return -1
int32_t raftCachePutEntry(struct SRaftEntryHashCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
if (pCache->currentCount >= pCache->maxCount) {
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosHashPut(pCache->pEntryHash, &(pEntry->index), sizeof(pEntry->index), pEntry, pEntry->bytes);
++(pCache->currentCount);
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
taosThreadMutexUnlock(&(pCache->mutex));
sNTrace(pCache->pSyncNode, "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
taosThreadMutexUnlock(&pCache->mutex);
return 1;
}
@ -269,26 +264,21 @@ int32_t raftCacheGetEntry(struct SRaftEntryHashCache* pCache, SyncIndex index, S
}
*ppEntry = NULL;
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = taosMemoryMalloc(pEntry->bytes);
memcpy(*ppEntry, pTmp, pEntry->bytes);
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
taosThreadMutexUnlock(&(pCache->mutex));
sNTrace(pCache->pSyncNode, "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
@ -302,34 +292,29 @@ int32_t raftCacheGetEntryP(struct SRaftEntryHashCache* pCache, SyncIndex index,
}
*ppEntry = NULL;
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = pEntry;
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
taosThreadMutexUnlock(&(pCache->mutex));
sNTrace(pCache->pSyncNode, "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
int32_t raftCacheDelEntry(struct SRaftEntryHashCache* pCache, SyncIndex index) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
taosHashRemove(pCache->pEntryHash, &index, sizeof(index));
--(pCache->currentCount);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
@ -339,38 +324,34 @@ int32_t raftCacheGetAndDel(struct SRaftEntryHashCache* pCache, SyncIndex index,
}
*ppEntry = NULL;
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = taosMemoryMalloc(pEntry->bytes);
memcpy(*ppEntry, pTmp, pEntry->bytes);
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache get-and-del, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
sNTrace(pCache->pSyncNode, "raft cache get-and-del, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosHashRemove(pCache->pEntryHash, &index, sizeof(index));
--(pCache->currentCount);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
int32_t raftCacheClear(struct SRaftEntryHashCache* pCache) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
taosHashClear(pCache->pEntryHash);
pCache->currentCount = 0;
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
@ -380,7 +361,7 @@ cJSON* raftCache2Json(SRaftEntryHashCache* pCache) {
cJSON* pRoot = cJSON_CreateObject();
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
snprintf(u64buf, sizeof(u64buf), "%p", pCache->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
@ -402,7 +383,7 @@ cJSON* raftCache2Json(SRaftEntryHashCache* pCache) {
}
}
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
}
cJSON* pJson = cJSON_CreateObject();
@ -483,13 +464,13 @@ SRaftEntryCache* raftEntryCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
void raftEntryCacheDestroy(SRaftEntryCache* pCache) {
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
tSkipListDestroy(pCache->pSkipList);
if (pCache->refMgr != -1) {
taosCloseRef(pCache->refMgr);
pCache->refMgr = -1;
}
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
taosThreadMutexDestroy(&(pCache->mutex));
taosMemoryFree(pCache);
}
@ -499,10 +480,10 @@ void raftEntryCacheDestroy(SRaftEntryCache* pCache) {
// max count, return 0
// error, return -1
int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
if (pCache->currentCount >= pCache->maxCount) {
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
@ -513,15 +494,10 @@ int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* p
pEntry->rid = taosAddRef(pCache->refMgr, pEntry);
ASSERT(pEntry->rid >= 0);
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
taosThreadMutexUnlock(&(pCache->mutex));
sNTrace(pCache->pSyncNode, "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
taosThreadMutexUnlock(&pCache->mutex);
return 1;
}
@ -546,7 +522,7 @@ int32_t raftEntryCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index,
// not found, return 0
// error, return -1
int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
SyncIndex index2 = index;
int32_t code = 0;
@ -570,7 +546,7 @@ int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index,
}
taosArrayDestroy(entryPArray);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return code;
}
@ -579,7 +555,7 @@ int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index,
// return -1, error
// return delete count
int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
int32_t returnCnt = 0;
if (count == -1) {
@ -634,7 +610,7 @@ int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count) {
}
pCache->currentCount -= returnCnt;
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
return returnCnt;
}
@ -643,7 +619,7 @@ cJSON* raftEntryCache2Json(SRaftEntryCache* pCache) {
cJSON* pRoot = cJSON_CreateObject();
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
taosThreadMutexLock(&pCache->mutex);
snprintf(u64buf, sizeof(u64buf), "%p", pCache->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
@ -661,7 +637,7 @@ cJSON* raftEntryCache2Json(SRaftEntryCache* pCache) {
}
tSkipListDestroyIter(pIter);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexUnlock(&pCache->mutex);
}
cJSON* pJson = cJSON_CreateObject();

54
source/libs/sync/src/syncRaftLog.c
View File

@ -117,12 +117,9 @@ static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncI
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"wal restore from snapshot error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
snapshotIndex, err, err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
sNError(pData->pSyncNode,
"wal restore from snapshot error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
snapshotIndex, err, err, errStr, sysErr, sysErrStr);
return -1;
}
@ -214,23 +211,14 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal write error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pEntry->index, err, err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
// ASSERT(0);
sNError(pData->pSyncNode, "wal write error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pEntry->index, err, err, errStr, sysErr, sysErrStr);
return -1;
}
pEntry->index = index;
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "write index:%" PRId64 ", type:%s, origin type:%s", pEntry->index,
TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
syncNodeEventLog(pData->pSyncNode, eventLog);
} while (0);
sNTrace(pData->pSyncNode, "write index:%" PRId64 ", type:%s, origin type:%s", pEntry->index,
TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
return 0;
}
@ -261,20 +249,13 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
do {
char logBuf[128];
if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
snprintf(logBuf, sizeof(logBuf),
"wal read not exist, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s", index, err, err,
errStr, sysErr, sysErrStr);
syncNodeEventLog(pData->pSyncNode, logBuf);
} else {
snprintf(logBuf, sizeof(logBuf), "wal read error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
index, err, err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
}
} while (0);
if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
sNTrace(pData->pSyncNode, "wal read not exist, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s", index,
err, err, errStr, sysErr, sysErrStr);
} else {
sNTrace(pData->pSyncNode, "wal read error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s", index,
err, err, errStr, sysErr, sysErrStr);
}
/*
int32_t saveErr = terrno;
@ -331,12 +312,7 @@ static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIn
}
// event log
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "log truncate, from-index:%" PRId64, fromIndex);
syncNodeEventLog(pData->pSyncNode, logBuf);
} while (0);
sNTrace(pData->pSyncNode, "log truncate, from-index:%" PRId64, fromIndex);
return code;
}

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

@ -56,14 +56,10 @@ int32_t syncNodeReplicateOne(SSyncNode* pSyncNode, SRaftId* pDestId) {
SyncIndex logStartIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
SyncIndex logEndIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
if (nextIndex < logStartIndex || nextIndex - 1 > logEndIndex) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "maybe start snapshot for next-index:%" PRId64 ", start:%" PRId64 ", end:%" PRId64,
nextIndex, logStartIndex, logEndIndex);
syncNodeEventLog(pSyncNode, logBuf);
sNTrace(pSyncNode, "maybe start snapshot for next-index:%" PRId64 ", start:%" PRId64 ", end:%" PRId64, nextIndex,
logStartIndex, logEndIndex);
// start snapshot
// int32_t code = syncNodeStartSnapshot(pSyncNode, pDestId);
return 0;
}
@ -103,10 +99,7 @@ int32_t syncNodeReplicateOne(SSyncNode* pSyncNode, SRaftId* pDestId) {
char host[64];
uint16_t port;
syncUtilU642Addr(pDestId->addr, host, sizeof(host), &port);
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "replicate to %s:%d error, next-index:%" PRId64, host, port, nextIndex);
syncNodeErrorLog(pSyncNode, logBuf);
sNError(pSyncNode, "replicate to %s:%d error, next-index:%" PRId64, host, port, nextIndex);
} while (0);
syncAppendEntriesDestroy(pMsg);
@ -137,7 +130,7 @@ int32_t syncNodeReplicate(SSyncNode* pSyncNode) {
return -1;
}
syncNodeEventLog(pSyncNode, "do replicate");
sNTrace(pSyncNode, "do replicate");
int32_t ret = 0;
for (int i = 0; i < pSyncNode->peersNum; ++i) {
@ -186,9 +179,7 @@ int32_t syncNodeMaybeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* dest
char host[64];
int16_t port;
syncUtilU642Addr(destRaftId->addr, host, sizeof(host), &port);
snprintf(logBuf, sizeof(logBuf), "do not repcate to %s:%d for index:%" PRId64, host, port, pMsg->prevLogIndex + 1);
syncNodeEventLog(pSyncNode, logBuf);
sNTrace(pSyncNode, "do not repcate to %s:%d for index:%" PRId64, host, port, pMsg->prevLogIndex + 1);
}
return ret;

64
source/libs/sync/src/syncRequestVote.c
View File

@ -48,66 +48,42 @@ static bool syncNodeOnRequestVoteLogOK(SSyncNode* pSyncNode, SyncRequestVote* pM
SyncIndex myLastIndex = syncNodeGetLastIndex(pSyncNode);
if (pMsg->lastLogIndex < pSyncNode->commitIndex) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode,
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
return false;
}
if (myLastTerm == SYNC_TERM_INVALID) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode,
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
return false;
}
if (pMsg->lastLogTerm > myLastTerm) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"logok:1, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode,
"logok:1, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
return true;
}
if (pMsg->lastLogTerm == myLastTerm && pMsg->lastLogIndex >= myLastIndex) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"logok:1, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode,
"logok:1, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
return true;
}
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
sNTrace(pSyncNode,
"logok:0, {my-lterm:%" PRIu64 ", my-lindex:%" PRId64 ", recv-lterm:%" PRIu64 ", recv-lindex:%" PRId64
", recv-term:%" PRIu64 "}",
myLastTerm, myLastIndex, pMsg->lastLogTerm, pMsg->lastLogIndex, pMsg->term);
return false;
}

6
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -71,10 +71,8 @@ int32_t syncNodeOnRequestVoteReply(SSyncNode* ths, SyncRequestVoteReply* pMsg) {
// but they won't be looked at, so it doesn't matter.
if (ths->state == TAOS_SYNC_STATE_CANDIDATE) {
if (ths->pVotesRespond->term != pMsg->term) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "vote respond error vote-respond-mgr term:%" PRIu64 ", msg term:%" PRIu64 "",
ths->pVotesRespond->term, pMsg->term);
syncNodeErrorLog(ths, logBuf);
sNError(ths, "vote respond error vote-respond-mgr term:%" PRIu64 ", msg term:%" PRIu64 "",
ths->pVotesRespond->term, pMsg->term);
return -1;
}

24
source/libs/sync/src/syncRespMgr.c
View File

@ -52,12 +52,8 @@ int64_t syncRespMgrAdd(SSyncRespMgr *pObj, SRespStub *pStub) {
uint64_t keyCode = ++(pObj->seqNum);
taosHashPut(pObj->pRespHash, &keyCode, sizeof(keyCode), pStub, sizeof(SRespStub));
SSyncNode *pSyncNode = pObj->data;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "save message handle, type:%s seq:%" PRIu64 " handle:%p",
TMSG_INFO(pStub->rpcMsg.msgType), keyCode, pStub->rpcMsg.info.handle);
syncNodeEventLog(pSyncNode, eventLog);
sNTrace(pObj->data, "save message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType),
keyCode, pStub->rpcMsg.info.handle);
taosThreadMutexUnlock(&(pObj->mutex));
return keyCode;
}
@ -78,12 +74,8 @@ int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub) {
if (pTmp != NULL) {
memcpy(pStub, pTmp, sizeof(SRespStub));
SSyncNode *pSyncNode = pObj->data;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "get message handle, type:%s seq:%" PRIu64 " handle:%p",
TMSG_INFO(pStub->rpcMsg.msgType), index, pStub->rpcMsg.info.handle);
syncNodeEventLog(pSyncNode, eventLog);
sNTrace(pObj->data, "get message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType),
index, pStub->rpcMsg.info.handle);
taosThreadMutexUnlock(&(pObj->mutex));
return 1; // get one object
}
@ -98,12 +90,8 @@ int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStu
if (pTmp != NULL) {
memcpy(pStub, pTmp, sizeof(SRespStub));
SSyncNode *pSyncNode = pObj->data;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "get-and-del message handle, type:%s seq:%" PRIu64 " handle:%p",
TMSG_INFO(pStub->rpcMsg.msgType), index, pStub->rpcMsg.info.handle);
syncNodeEventLog(pSyncNode, eventLog);
sNTrace(pObj->data, "get-and-del message handle, type:%s seq:%" PRIu64 " handle:%p",
TMSG_INFO(pStub->rpcMsg.msgType), index, pStub->rpcMsg.info.handle);
taosHashRemove(pObj->pRespHash, &index, sizeof(index));
taosThreadMutexUnlock(&(pObj->mutex));
return 1; // get one object

207
source/libs/sync/src/syncSnapshot.c
View File

@ -73,7 +73,7 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
if (pSender->pReader != NULL) {
int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader);
if (ret != 0) {
syncNodeErrorLog(pSender->pSyncNode, "stop reader error");
sNError(pSender->pSyncNode, "stop reader error");
}
pSender->pReader = NULL;
}
@ -130,12 +130,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender) {
syncSnapshotSendDestroy(pMsg);
// event log
do {
char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender start");
syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSTrace(pSender, "snapshot sender start");
return 0;
}
@ -159,12 +154,7 @@ int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
}
// event log
do {
char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender stop");
syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSTrace(pSender, "snapshot sender stop");
return 0;
}
@ -212,17 +202,11 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
syncSnapshotSendDestroy(pMsg);
// event log
do {
char *eventLog = NULL;
if (pSender->seq == SYNC_SNAPSHOT_SEQ_END) {
eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender finish");
} else {
eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender sending");
}
syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
if (pSender->seq == SYNC_SNAPSHOT_SEQ_END) {
sSTrace(pSender, "snapshot sender finish");
} else {
sSTrace(pSender, "snapshot sender sending");
}
return 0;
}
@ -253,11 +237,7 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) {
syncSnapshotSendDestroy(pMsg);
// event log
do {
char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender resend");
syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSTrace(pSender, "snapshot sender resend");
}
return 0;
@ -287,10 +267,10 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
if (pSender->pCurrentBlock != NULL) {
char *s;
s = syncUtilprintBin((char *)(pSender->pCurrentBlock), pSender->blockLen);
s = syncUtilPrintBin((char *)(pSender->pCurrentBlock), pSender->blockLen);
cJSON_AddStringToObject(pRoot, "pCurrentBlock", s);
taosMemoryFree(s);
s = syncUtilprintBin2((char *)(pSender->pCurrentBlock), pSender->blockLen);
s = syncUtilPrintBin2((char *)(pSender->pCurrentBlock), pSender->blockLen);
cJSON_AddStringToObject(pRoot, "pCurrentBlock2", s);
taosMemoryFree(s);
}
@ -327,31 +307,12 @@ char *snapshotSender2Str(SSyncSnapshotSender *pSender) {
return serialized;
}
char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
int32_t len = 256;
char *s = taosMemoryMalloc(len);
SRaftId destId = pSender->pSyncNode->replicasId[pSender->replicaIndex];
char host[64];
uint16_t port;
syncUtilU642Addr(destId.addr, host, sizeof(host), &port);
snprintf(s, len,
"%s {%p s-param:%" PRId64 " e-param:%" PRId64 " laindex:%" PRId64 " laterm:%" PRIu64 " lcindex:%" PRId64
" seq:%d ack:%d finish:%d replica-index:%d %s:%d}",
event, pSender, pSender->snapshotParam.start, pSender->snapshotParam.end, pSender->snapshot.lastApplyIndex,
pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack,
pSender->finish, pSender->replicaIndex, host, port);
return s;
}
int32_t syncNodeStartSnapshot(SSyncNode *pSyncNode, SRaftId *pDestId) {
syncNodeEventLog(pSyncNode, "starting snapshot ...");
sNTrace(pSyncNode, "starting snapshot ...");
SSyncSnapshotSender *pSender = syncNodeGetSnapshotSender(pSyncNode, pDestId);
if (pSender == NULL) {
syncNodeErrorLog(pSyncNode, "start snapshot error, sender is null");
sNError(pSyncNode, "start snapshot error, sender is null");
return -1;
}
@ -360,14 +321,14 @@ int32_t syncNodeStartSnapshot(SSyncNode *pSyncNode, SRaftId *pDestId) {
if (snapshotSenderIsStart(pSender)) {
code = snapshotSenderStop(pSender, false);
if (code != 0) {
syncNodeErrorLog(pSyncNode, "snapshot sender stop error");
sNError(pSyncNode, "snapshot sender stop error");
return -1;
}
}
code = snapshotSenderStart(pSender);
if (code != 0) {
syncNodeErrorLog(pSyncNode, "snapshot sender start error");
sNError(pSyncNode, "snapshot sender start error");
return -1;
}
@ -447,11 +408,7 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapsh
pReceiver->startTime = pBeginMsg->startTime;
// event log
do {
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver start");
syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver start");
}
// force stop
@ -467,11 +424,7 @@ void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
pReceiver->start = false;
// event log
do {
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver force stop");
syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver force stop");
}
// if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
@ -495,14 +448,7 @@ int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
pReceiver->start = false;
// event log
do {
SSnapshot snapshot;
pReceiver->pSyncNode->pFsm->FpGetSnapshotInfo(pReceiver->pSyncNode->pFsm, &snapshot);
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver stop");
syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver stop");
return 0;
}
@ -517,7 +463,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data,
pMsg->dataLen);
if (code != 0) {
syncNodeErrorLog(pReceiver->pSyncNode, "snapshot write error");
sNError(pReceiver->pSyncNode, "snapshot write error");
return -1;
}
}
@ -526,7 +472,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
code =
pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
if (code != 0) {
syncNodeErrorLog(pReceiver->pSyncNode, "wal restore from snapshot error");
sNError(pReceiver->pSyncNode, "wal restore from snapshot error");
return -1;
}
@ -545,8 +491,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true,
&(pReceiver->snapshot));
if (code != 0) {
syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
// ASSERT(0);
sNError(pReceiver->pSyncNode, "snapshot stop writer true error");
return -1;
}
pReceiver->pWriter = NULL;
@ -555,19 +500,12 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
} else {
syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
sNError(pReceiver->pSyncNode, "snapshot stop writer true error");
return -1;
}
// event log
do {
SSnapshot snapshot;
pReceiver->pSyncNode->pFsm->FpGetSnapshotInfo(pReceiver->pSyncNode->pFsm, &snapshot);
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver got last data, finish, apply snapshot");
syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver got last data, finish, apply snapshot");
return 0;
}
@ -588,11 +526,7 @@ static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapsh
pReceiver->ack = pMsg->seq;
// event log
do {
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver receiving");
syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver receiving");
}
}
@ -653,27 +587,6 @@ char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
return serialized;
}
char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) {
int32_t len = 256;
char *s = taosMemoryMalloc(len);
SRaftId fromId = pReceiver->fromId;
char host[128];
uint16_t port;
syncUtilU642Addr(fromId.addr, host, sizeof(host), &port);
snprintf(s, len,
"%s {%p start:%d ack:%d term:%" PRIu64 " start-time:%" PRId64 " from:%s:%d s-param:%" PRId64
" e-param:%" PRId64 " laindex:%" PRId64 " laterm:%" PRIu64
" "
"lcindex:%" PRId64 "}",
event, pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->startTime, host, port,
pReceiver->snapshotParam.start, pReceiver->snapshotParam.end, pReceiver->snapshot.lastApplyIndex,
pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex);
return s;
}
SyncIndex syncNodeGetSnapBeginIndex(SSyncNode *ths) {
SyncIndex snapStart = SYNC_INDEX_INVALID;
@ -688,11 +601,8 @@ SyncIndex syncNodeGetSnapBeginIndex(SSyncNode *ths) {
int64_t walCommitVer = walGetCommittedVer(pWal);
if (!isEmpty && ths->commitIndex != walCommitVer) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "commit not same, wal-commit:%" PRId64 ", commit:%" PRId64 ", ignore",
walCommitVer, ths->commitIndex);
syncNodeErrorLog(ths, logBuf);
sNError(ths, "commit not same, wal-commit:%" PRId64 ", commit:%" PRId64 ", ignore", walCommitVer,
ths->commitIndex);
snapStart = walCommitVer + 1;
} else {
snapStart = ths->commitIndex + 1;
@ -726,7 +636,7 @@ static int32_t syncNodeOnSnapshotPre(SSyncNode *pSyncNode, SyncSnapshotSend *pMs
_START_RECEIVER:
if (taosGetTimestampMs() - pMsg->startTime > SNAPSHOT_MAX_CLOCK_SKEW_MS) {
syncNodeErrorLog(pSyncNode, "snapshot receiver time skew too much");
sNError(pSyncNode, "snapshot receiver time skew too much");
return -1;
} else {
// waiting for clock match
@ -772,14 +682,14 @@ static int32_t syncNodeOnSnapshotBegin(SSyncNode *pSyncNode, SyncSnapshotSend *p
return 0;
} else {
// ignore
syncNodeEventLog(pSyncNode, "msg ignore");
sNTrace(pSyncNode, "msg ignore");
return 0;
}
}
_START_RECEIVER:
if (taosGetTimestampMs() - pMsg->startTime > SNAPSHOT_MAX_CLOCK_SKEW_MS) {
syncNodeErrorLog(pSyncNode, "snapshot receiver time skew too much");
sNError(pSyncNode, "snapshot receiver time skew too much");
return -1;
} else {
// waiting for clock match
@ -899,35 +809,18 @@ int32_t syncNodeOnSnapshot(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
} else {
// error log
do {
char logBuf[96];
snprintf(logBuf, sizeof(logBuf), "snapshot receiver recv error seq:%d, my ack:%d", pMsg->seq, pReceiver->ack);
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, logBuf);
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver recv error seq:%d, my ack:%d", pMsg->seq, pReceiver->ack);
return -1;
}
} else {
// error log
do {
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver term not equal");
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver term not equal");
return -1;
}
} else {
// error log
do {
char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver not follower");
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sRTrace(pReceiver, "snapshot receiver not follower");
return -1;
}
@ -947,14 +840,14 @@ int32_t syncNodeOnSnapshotReplyPre(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg)
pSender->snapshotParam.end = snapshot.lastApplyIndex;
if (pMsg->snapBeginIndex > snapshot.lastApplyIndex) {
syncNodeErrorLog(pSyncNode, "snapshot last index too small");
sNError(pSyncNode, "snapshot last index too small");
return -1;
}
// start reader
int32_t code = pSyncNode->pFsm->FpSnapshotStartRead(pSyncNode->pFsm, &(pSender->snapshotParam), &(pSender->pReader));
if (code != 0) {
syncNodeErrorLog(pSyncNode, "create snapshot reader error");
sNError(pSyncNode, "create snapshot reader error");
return -1;
}
@ -1029,34 +922,17 @@ int32_t syncNodeOnSnapshotReply(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
} else {
// error log
do {
char logBuf[96];
snprintf(logBuf, sizeof(logBuf), "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);
char *eventLog = snapshotSender2SimpleStr(pSender, logBuf);
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSError(pSender, "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);
return -1;
}
} else {
// error log
do {
char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender term not equal");
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSError(pSender, "snapshot sender term not equal");
return -1;
}
} else {
// error log
do {
char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender not leader");
syncNodeErrorLog(pSyncNode, eventLog);
taosMemoryFree(eventLog);
} while (0);
sSError(pSender, "snapshot sender not leader");
return -1;
}
@ -1082,11 +958,8 @@ int32_t syncNodeOnPreSnapshot(SSyncNode *ths, SyncPreSnapshot *pMsg) {
int64_t walCommitVer = walGetCommittedVer(pWal);
if (!isEmpty && ths->commitIndex != walCommitVer) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "commit not same, wal-commit:%" PRId64 ", commit:%" PRId64 ", ignore",
walCommitVer, ths->commitIndex);
syncNodeErrorLog(ths, logBuf);
sNError(ths, "commit not same, wal-commit:%" PRId64 ", commit:%" PRId64 ", ignore", walCommitVer,
ths->commitIndex);
goto _IGNORE;
}
@ -1095,7 +968,7 @@ int32_t syncNodeOnPreSnapshot(SSyncNode *ths, SyncPreSnapshot *pMsg) {
// make local log clean
int32_t code = ths->pLogStore->syncLogTruncate(ths->pLogStore, pMsgReply->snapStart);
if (code != 0) {
syncNodeErrorLog(ths, "truncate wal error");
sNError(ths, "truncate wal error");
goto _IGNORE;
}
}

16
source/libs/sync/src/syncTimeout.c
View File

@ -49,17 +49,12 @@ static void syncNodeCleanConfigIndex(SSyncNode* ths) {
int32_t code = raftCfgPersist(ths->pRaftCfg);
ASSERT(code == 0);
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "clean config index arr, old-cnt:%d, new-cnt:%d", oldCnt,
ths->pRaftCfg->configIndexCount);
syncNodeEventLog(ths, logBuf);
} while (0);
sNTrace(ths, "clean config index arr, old-cnt:%d, new-cnt:%d", oldCnt, ths->pRaftCfg->configIndexCount);
}
}
int32_t syncNodeTimerRoutine(SSyncNode* ths) {
syncNodeEventLog(ths, "timer routines");
sNTrace(ths, "timer routines");
// timer replicate
syncNodeReplicate(ths);
@ -79,12 +74,7 @@ int32_t syncNodeTimerRoutine(SSyncNode* ths) {
sError("vgId:%d, timer wal snapshot end error since:%s", ths->vgId, terrstr());
return -1;
} else {
do {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "wal snapshot end, index:%" PRId64, atomic_load_64(&ths->snapshottingIndex));
syncNodeEventLog(ths, logBuf);
} while (0);
sNTrace(ths, "wal snapshot end, index:%" PRId64, atomic_load_64(&ths->snapshottingIndex));
atomic_store_64(&ths->snapshottingIndex, SYNC_INDEX_INVALID);
}
}

267
source/libs/sync/src/syncUtil.c
View File

@ -13,17 +13,17 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncUtil.h"
#include <stdio.h>
#include "syncEnv.h"
#include "syncInt.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
#include "syncSnapshot.h"
void addEpIntoEpSet(SEpSet* pEpSet, const char* fqdn, uint16_t port);
extern void addEpIntoEpSet(SEpSet* pEpSet, const char* fqdn, uint16_t port);
// ---- encode / decode
uint64_t syncUtilAddr2U64(const char* host, uint16_t port) {
uint64_t u64;
uint32_t hostU32 = taosGetIpv4FromFqdn(host);
if (hostU32 == (uint32_t)-1) {
sError("failed to resolve ipv4 addr. host:%s", host);
@ -31,41 +31,22 @@ uint64_t syncUtilAddr2U64(const char* host, uint16_t port) {
return -1;
}
/*
uint32_t hostU32 = (uint32_t)taosInetAddr(host);
if (hostU32 == (uint32_t)-1) {
struct hostent* hostEnt = gethostbyname(host);
if (hostEnt == NULL) {
sError("Get IP address error");
return -1;
}
const char* newHost = taosInetNtoa(*(struct in_addr*)(hostEnt->h_addr_list[0]));
hostU32 = (uint32_t)taosInetAddr(newHost);
if (hostU32 == (uint32_t)-1) {
sError("change %s to id, error", newHost);
}
// ASSERT(hostU32 != (uint32_t)-1);
}
*/
u64 = (((uint64_t)hostU32) << 32) | (((uint32_t)port) << 16);
uint64_t u64 = (((uint64_t)hostU32) << 32) | (((uint32_t)port) << 16);
return u64;
}
void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port) {
void syncUtilU642Addr(uint64_t u64, char* host, int64_t len, uint16_t* port) {
uint32_t hostU32 = (uint32_t)((u64 >> 32) & 0x00000000FFFFFFFF);
struct in_addr addr;
addr.s_addr = hostU32;
struct in_addr addr = {.s_addr = hostU32};
taosInetNtoa(addr, host, len);
*port = (uint16_t)((u64 & 0x00000000FFFF0000) >> 16);
}
void syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet) {
void syncUtilnodeInfo2EpSet(const SNodeInfo* pInfo, SEpSet* pEpSet) {
pEpSet->inUse = 0;
pEpSet->numOfEps = 0;
addEpIntoEpSet(pEpSet, pNodeInfo->nodeFqdn, pNodeInfo->nodePort);
addEpIntoEpSet(pEpSet, pInfo->nodeFqdn, pInfo->nodePort);
}
void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet) {
@ -73,28 +54,22 @@ void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet) {
uint16_t port;
syncUtilU642Addr(raftId->addr, host, sizeof(host), &port);
/*
pEpSet->numOfEps = 1;
pEpSet->inUse = 0;
pEpSet->eps[0].port = port;
snprintf(pEpSet->eps[0].fqdn, sizeof(pEpSet->eps[0].fqdn), "%s", host);
*/
pEpSet->inUse = 0;
pEpSet->numOfEps = 0;
addEpIntoEpSet(pEpSet, host, port);
}
bool syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId) {
uint32_t ipv4 = taosGetIpv4FromFqdn(pNodeInfo->nodeFqdn);
bool syncUtilnodeInfo2raftId(const SNodeInfo* pInfo, SyncGroupId vgId, SRaftId* raftId) {
uint32_t ipv4 = taosGetIpv4FromFqdn(pInfo->nodeFqdn);
if (ipv4 == 0xFFFFFFFF || ipv4 == 1) {
sError("failed to resolve ipv4 addr. fqdn: %s", pNodeInfo->nodeFqdn);
sError("failed to resolve ipv4 addr. fqdn: %s", pInfo->nodeFqdn);
terrno = TSDB_CODE_TSC_INVALID_FQDN;
return false;
}
char ipbuf[128] = {0};
tinet_ntoa(ipbuf, ipv4);
raftId->addr = syncUtilAddr2U64(ipbuf, pNodeInfo->nodePort);
raftId->addr = syncUtilAddr2U64(ipbuf, pInfo->nodePort);
raftId->vgId = vgId;
return true;
}
@ -106,31 +81,9 @@ bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2) {
bool syncUtilEmptyId(const SRaftId* pId) { return (pId->addr == 0 && pId->vgId == 0); }
// ---- SSyncBuffer -----
void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len) {
syncBuf->len = len;
syncBuf->data = taosMemoryMalloc(syncBuf->len);
}
void syncUtilbufDestroy(SSyncBuffer* syncBuf) { taosMemoryFree(syncBuf->data); }
void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest) {
dest->len = src->len;
dest->data = src->data;
}
void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest) {
dest->len = src->len;
dest->data = taosMemoryMalloc(dest->len);
memcpy(dest->data, src->data, dest->len);
}
// ---- misc ----
int32_t syncUtilRand(int32_t max) { return taosRand() % max; }
static inline int32_t syncUtilRand(int32_t max) { return taosRand() % max; }
int32_t syncUtilElectRandomMS(int32_t min, int32_t max) {
ASSERT(min > 0 && max > 0 && max >= min);
int32_t rdm = min + syncUtilRand(max - min);
// sDebug("random min:%d, max:%d, rdm:%d", min, max, rdm);
@ -176,7 +129,7 @@ char* syncUtilRaftId2Str(const SRaftId* p) {
return serialized;
}
bool syncUtilCanPrint(char c) {
static inline bool syncUtilCanPrint(char c) {
if (c >= 32 && c <= 126) {
return true;
} else {
@ -184,7 +137,7 @@ bool syncUtilCanPrint(char c) {
}
}
char* syncUtilprintBin(char* ptr, uint32_t len) {
char* syncUtilPrintBin(char* ptr, uint32_t len) {
int64_t memLen = (int64_t)(len + 1);
char* s = taosMemoryMalloc(memLen);
ASSERT(s != NULL);
@ -199,7 +152,7 @@ char* syncUtilprintBin(char* ptr, uint32_t len) {
return s;
}
char* syncUtilprintBin2(char* ptr, uint32_t len) {
char* syncUtilPrintBin2(char* ptr, uint32_t len) {
uint32_t len2 = len * 4 + 1;
char* s = taosMemoryMalloc(len2);
ASSERT(s != NULL);
@ -213,16 +166,6 @@ char* syncUtilprintBin2(char* ptr, uint32_t len) {
return s;
}
SyncIndex syncUtilMinIndex(SyncIndex a, SyncIndex b) {
SyncIndex r = a < b ? a : b;
return r;
}
SyncIndex syncUtilMaxIndex(SyncIndex a, SyncIndex b) {
SyncIndex r = a > b ? a : b;
return r;
}
void syncUtilMsgHtoN(void* msg) {
// htonl
SMsgHead* pHead = msg;
@ -237,15 +180,6 @@ void syncUtilMsgNtoH(void* msg) {
pHead->vgId = ntohl(pHead->vgId);
}
#if 0
bool syncUtilIsData(tmsg_t msgType) {
if (msgType == TDMT_SYNC_NOOP || msgType == TDMT_SYNC_CONFIG_CHANGE) {
return false;
}
return true;
}
#endif
bool syncUtilUserPreCommit(tmsg_t msgType) {
if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH &&
msgType != TDMT_SYNC_LEADER_TRANSFER) {
@ -273,24 +207,149 @@ bool syncUtilUserRollback(tmsg_t msgType) {
return false;
}
void syncUtilJson2Line(char* jsonStr) {
int p, q, len;
p = 0;
q = 1;
len = strlen(jsonStr);
while (1) {
if (jsonStr[q] == '\0') {
jsonStr[p + 1] = '\0';
break;
}
void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) {
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
if (jsonStr[q] == '\n' || jsonStr[q] == ' ' || jsonStr[q] == '\t') {
q++;
continue;
} else {
jsonStr[p + 1] = jsonStr[q];
p++;
q++;
}
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);
}
SyncIndex logLastIndex = SYNC_INDEX_INVALID;
SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
if (pNode->pLogStore != NULL) {
logLastIndex = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
logBeginIndex = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
}
char cfgStr[1024];
syncCfg2SimpleStr(&(pNode->pRaftCfg->cfg), cfgStr, sizeof(cfgStr));
char* pPeerStateStr = syncNodePeerState2Str(pNode);
int32_t quorum = syncNodeDynamicQuorum(pNode);
char eventLog[512]; // {0};
va_list argpointer;
va_start(argpointer, format);
int32_t writeLen = vsnprintf(eventLog, sizeof(eventLog), format, argpointer);
va_end(argpointer);
taosPrintLog(flags, level, dflag,
"vgId:%d, sync %s "
"%s"
", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
", snap-tm:%" PRIu64 ", sby:%d, stgy:%d, bch:%d, r-num:%d, lcfg:%" PRId64
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
pNode->vgId, syncStr(pNode->state), eventLog, pNode->pRaftStore->currentTerm, pNode->commitIndex,
logBeginIndex, logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
pNode->pRaftCfg->isStandBy, pNode->pRaftCfg->snapshotStrategy, pNode->pRaftCfg->batchSize,
pNode->replicaNum, pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, pPeerStateStr, cfgStr);
taosMemoryFree(pPeerStateStr);
}
void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender,
const char* format, ...) {
SSyncNode* pNode = pSender->pSyncNode;
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);
}
SyncIndex logLastIndex = SYNC_INDEX_INVALID;
SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
if (pNode->pLogStore != NULL) {
logLastIndex = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
logBeginIndex = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
}
char cfgStr[1024];
syncCfg2SimpleStr(&(pNode->pRaftCfg->cfg), cfgStr, sizeof(cfgStr));
char* pPeerStateStr = syncNodePeerState2Str(pNode);
int32_t quorum = syncNodeDynamicQuorum(pNode);
SRaftId destId = pNode->replicasId[pSender->replicaIndex];
char host[64];
uint16_t port;
syncUtilU642Addr(destId.addr, host, sizeof(host), &port);
char eventLog[512]; // {0};
va_list argpointer;
va_start(argpointer, format);
int32_t writeLen = vsnprintf(eventLog, sizeof(eventLog), format, argpointer);
va_end(argpointer);
taosPrintLog(flags, level, dflag,
"vgId:%d, sync %s "
"%s {%p s-param:%" PRId64 " e-param:%" PRId64 " laindex:%" PRId64 " laterm:%" PRIu64 " lcindex:%" PRId64
" seq:%d ack:%d finish:%d replica-index:%d %s:%d}"
", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
", snap-tm:%" PRIu64 ", sby:%d, stgy:%d, bch:%d, r-num:%d, lcfg:%" PRId64
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
pNode->vgId, syncStr(pNode->state), eventLog, pSender, pSender->snapshotParam.start,
pSender->snapshotParam.end, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm,
pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack, pSender->finish, pSender->replicaIndex,
host, port, pNode->pRaftStore->currentTerm, pNode->commitIndex, logBeginIndex, logLastIndex,
pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm, pNode->pRaftCfg->isStandBy,
pNode->pRaftCfg->snapshotStrategy, pNode->pRaftCfg->batchSize, pNode->replicaNum,
pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, pPeerStateStr, cfgStr);
taosMemoryFree(pPeerStateStr);
}
void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver,
const char* format, ...) {
SSyncNode* pNode = pReceiver->pSyncNode;
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);
}
SyncIndex logLastIndex = SYNC_INDEX_INVALID;
SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
if (pNode->pLogStore != NULL) {
logLastIndex = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
logBeginIndex = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
}
char cfgStr[1024];
syncCfg2SimpleStr(&(pNode->pRaftCfg->cfg), cfgStr, sizeof(cfgStr));
char* pPeerStateStr = syncNodePeerState2Str(pNode);
int32_t quorum = syncNodeDynamicQuorum(pNode);
SRaftId fromId = pReceiver->fromId;
char host[128];
uint16_t port;
syncUtilU642Addr(fromId.addr, host, sizeof(host), &port);
char eventLog[512]; // {0};
va_list argpointer;
va_start(argpointer, format);
int32_t writeLen = vsnprintf(eventLog, sizeof(eventLog), format, argpointer);
va_end(argpointer);
taosPrintLog(flags, level, dflag,
"vgId:%d, sync %s "
"%s {%p start:%d ack:%d term:%" PRIu64 " start-time:%" PRId64 " from:%s:%d s-param:%" PRId64
" e-param:%" PRId64 " laindex:%" PRId64 " laterm:%" PRIu64 " lcindex:%" PRId64
"}"
", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
", snap-tm:%" PRIu64 ", sby:%d, stgy:%d, bch:%d, r-num:%d, lcfg:%" PRId64
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
pNode->vgId, syncStr(pNode->state), eventLog, pReceiver, pReceiver->start, pReceiver->ack,
pReceiver->term, pReceiver->startTime, host, port, pReceiver->snapshotParam.start,
pReceiver->snapshotParam.end, pReceiver->snapshot.lastApplyIndex, pReceiver->snapshot.lastApplyTerm,
pReceiver->snapshot.lastConfigIndex, pNode->pRaftStore->currentTerm, pNode->commitIndex, logBeginIndex,
logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
pNode->pRaftCfg->isStandBy, pNode->pRaftCfg->snapshotStrategy, pNode->pRaftCfg->batchSize,
pNode->replicaNum, pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, pPeerStateStr, cfgStr);
taosMemoryFree(pPeerStateStr);
}

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

@ -68,7 +68,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) {
ASSERT(pMsg->voteGranted == true);
if (pMsg->term != pVotesGranted->term) {
syncNodeEventLog(pVotesGranted->pSyncNode, "vote grant vnode error");
sNTrace(pVotesGranted->pSyncNode, "vote grant vnode error");
return;
}
@ -207,7 +207,7 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) {
void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg) {
if (pVotesRespond->term != pMsg->term) {
syncNodeEventLog(pVotesRespond->pSyncNode, "vote respond add error");
sNTrace(pVotesRespond->pSyncNode, "vote respond add error");
return;
}

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

@ -191,7 +191,7 @@ int main(int argc, char **argv) {
// step5
uint32_t len;
char *pMsg5 = step5(pMsg4, &len);
char *s = syncUtilprintBin(pMsg5, len);
char *s = syncUtilPrintBin(pMsg5, len);
printf("==step5== [%s] \n", s);
taosMemoryFree(s);

5
source/libs/sync/test/syncRaftCfgIndexTest.cpp
View File

@ -60,7 +60,6 @@ void test1() {
ASSERT(code == 0);
SRaftCfgIndex* pRaftCfgIndex = raftCfgIndexOpen(pFile);
raftCfgIndexLog2((char*)"==test1==", pRaftCfgIndex);
raftCfgIndexClose(pRaftCfgIndex);
}
@ -71,15 +70,11 @@ void test2() {
raftCfgIndexAddConfigIndex(pRaftCfgIndex, i);
}
raftCfgIndexPersist(pRaftCfgIndex);
raftCfgIndexLog2((char*)"==test2==", pRaftCfgIndex);
raftCfgIndexClose(pRaftCfgIndex);
}
void test3() {
SRaftCfgIndex* pRaftCfgIndex = raftCfgIndexOpen(pFile);
raftCfgIndexLog2((char*)"==test3==", pRaftCfgIndex);
raftCfgIndexClose(pRaftCfgIndex);
}

12
source/libs/sync/test/syncRaftCfgTest.cpp
View File

@ -53,14 +53,6 @@ SSyncCfg* createSyncCfg() {
return pCfg;
}
void test1() {
SSyncCfg* pCfg = createSyncCfg();
syncCfgLog2((char*)__FUNCTION__, pCfg);
syncCfgLog3((char*)__FUNCTION__, pCfg);
taosMemoryFree(pCfg);
}
void test2() {
SSyncCfg* pCfg = createSyncCfg();
char* s = syncCfg2Str(pCfg);
@ -68,7 +60,6 @@ void test2() {
SSyncCfg* pCfg2 = (SSyncCfg*)taosMemoryMalloc(sizeof(SSyncCfg));
int32_t ret = syncCfgFromStr(s, pCfg2);
assert(ret == 0);
syncCfgLog2((char*)__FUNCTION__, pCfg2);
taosMemoryFree(pCfg);
taosMemoryFree(s);
@ -98,8 +89,6 @@ void test4() {
SRaftCfg* pCfg = raftCfgOpen("./test3_raft_cfg.json");
assert(pCfg != NULL);
raftCfgLog2((char*)__FUNCTION__, pCfg);
int32_t ret = raftCfgClose(pCfg);
assert(ret == 0);
}
@ -135,7 +124,6 @@ int main() {
sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
logTest();
test1();
test2();
test3();
test4();

22
source/libs/sync/test/syncTimeoutTest.cpp
View File

@ -78,6 +78,28 @@ void test5() {
syncTimeoutDestroy(pMsg2);
}
void syncUtilJson2Line(char *jsonStr) {
int p, q, len;
p = 0;
q = 1;
len = strlen(jsonStr);
while (1) {
if (jsonStr[q] == '\0') {
jsonStr[p + 1] = '\0';
break;
}
if (jsonStr[q] == '\n' || jsonStr[q] == ' ' || jsonStr[q] == '\t') {
q++;
continue;
} else {
jsonStr[p + 1] = jsonStr[q];
p++;
q++;
}
}
}
void test6() {
SyncTimeout *pMsg = createMsg();
char *jsonStr = syncTimeout2Str(pMsg);