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

enh: refact syncEnv code

This commit is contained in:
Shengliang Guan 2022-10-31 23:40:43 +08:00
parent 8cdbcb89f5
commit 392564cc7d
23 changed files with 224 additions and 364 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -200,10 +200,10 @@ typedef struct SSyncInfo {
int32_t syncInit(); int32_t syncInit();
void syncCleanUp(); void syncCleanUp();
bool syncIsInit();
int64_t syncOpen(SSyncInfo* pSyncInfo); int64_t syncOpen(SSyncInfo* pSyncInfo);
void syncStart(int64_t rid); void syncStart(int64_t rid);
void syncStop(int64_t rid); void syncStop(int64_t rid);
int32_t syncSetStandby(int64_t rid);
ESyncState syncGetMyRole(int64_t rid); ESyncState syncGetMyRole(int64_t rid);
bool syncIsReady(int64_t rid); bool syncIsReady(int64_t rid);
const char* syncGetMyRoleStr(int64_t rid); const char* syncGetMyRoleStr(int64_t rid);
@ -216,7 +216,6 @@ void syncGetEpSet(int64_t rid, SEpSet* pEpSet);
void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet); void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);
int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak); int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak);
// int32_t syncProposeBatch(int64_t rid, SRpcMsg** pMsgPArr, bool* pIsWeakArr, int32_t arrSize); // int32_t syncProposeBatch(int64_t rid, SRpcMsg** pMsgPArr, bool* pIsWeakArr, int32_t arrSize);
bool syncEnvIsStart();
const char* syncStr(ESyncState state); const char* syncStr(ESyncState state);
bool syncIsRestoreFinish(int64_t rid); bool syncIsRestoreFinish(int64_t rid);
int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot); int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot);
@ -234,6 +233,9 @@ int32_t syncEndSnapshot(int64_t rid);
int32_t syncStepDown(int64_t rid, SyncTerm newTerm); int32_t syncStepDown(int64_t rid, SyncTerm newTerm);
SSyncNode* syncNodeAcquire(int64_t rid);
void syncNodeRelease(SSyncNode* pNode);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

4
include/libs/sync/syncTools.h
View File

@ -28,10 +28,6 @@ typedef struct SRaftId {
SyncGroupId vgId; SyncGroupId vgId;
} SRaftId; } SRaftId;
// ------------------ control -------------------
SSyncNode* syncNodeAcquire(int64_t rid);
void syncNodeRelease(SSyncNode* pNode);
int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg); int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg);
int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo); int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo);
void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb); void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb);

3
include/util/tref.h
View File

@ -25,7 +25,8 @@ extern "C" {
// open a reference set, max is the mod used by hash, fp is the pointer to free resource function // open a reference set, max is the mod used by hash, fp is the pointer to free resource function
// return rsetId which will be used by other APIs. On error, -1 is returned, and terrno is set appropriately // return rsetId which will be used by other APIs. On error, -1 is returned, and terrno is set appropriately
int32_t taosOpenRef(int32_t max, void (*fp)(void *)); typedef void (*RefFp)(void *);
int32_t taosOpenRef(int32_t max, RefFp fp);
// close the reference set, refId is the return value by taosOpenRef // close the reference set, refId is the return value by taosOpenRef
// return 0 if success. On error, -1 is returned, and terrno is set appropriately // return 0 if success. On error, -1 is returned, and terrno is set appropriately

9
source/dnode/mnode/impl/src/mndMain.c
View File

@ -481,7 +481,7 @@ int32_t mndProcessSyncCtrlMsg(SRpcMsg *pMsg) {
mInfo("vgId:%d, process sync ctrl msg", 1); mInfo("vgId:%d, process sync ctrl msg", 1);
if (!syncEnvIsStart()) { if (!syncIsInit()) {
mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -518,7 +518,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
int32_t code = 0; int32_t code = 0;
if (!syncEnvIsStart()) { if (!syncIsInit()) {
mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -581,11 +581,6 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
code = syncNodeOnSnapshotReply(pSyncNode, pSyncMsg); code = syncNodeOnSnapshotReply(pSyncNode, pSyncMsg);
syncSnapshotRspDestroy(pSyncMsg); syncSnapshotRspDestroy(pSyncMsg);
} else if (pMsg->msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
code = syncSetStandby(pMgmt->sync);
SRpcMsg rsp = {.code = code, .info = pMsg->info};
tmsgSendRsp(&rsp);
} else if (pMsg->msgType == TDMT_SYNC_LOCAL_CMD) { } else if (pMsg->msgType == TDMT_SYNC_LOCAL_CMD) {
SyncLocalCmd *pSyncMsg = syncLocalCmdFromRpcMsg2(pMsg); SyncLocalCmd *pSyncMsg = syncLocalCmdFromRpcMsg2(pMsg);
code = syncNodeOnLocalCmd(pSyncNode, pSyncMsg); code = syncNodeOnLocalCmd(pSyncNode, pSyncMsg);

4
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -234,7 +234,7 @@ int32_t vnodeProcessSyncCtrlMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
int32_t code = 0; int32_t code = 0;
const STraceId *trace = &pMsg->info.traceId; const STraceId *trace = &pMsg->info.traceId;
if (!syncEnvIsStart()) { if (!syncIsInit()) {
vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg); vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg);
terrno = TSDB_CODE_APP_ERROR; terrno = TSDB_CODE_APP_ERROR;
return -1; return -1;
@ -277,7 +277,7 @@ int32_t vnodeProcessSyncMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
int32_t code = 0; int32_t code = 0;
const STraceId *trace = &pMsg->info.traceId; const STraceId *trace = &pMsg->info.traceId;
if (!syncEnvIsStart()) { if (!syncIsInit()) {
vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg); vGError("vgId:%d, msg:%p failed to process since sync env not start", pVnode->config.vgId, pMsg);
terrno = TSDB_CODE_APP_ERROR; terrno = TSDB_CODE_APP_ERROR;
return -1; return -1;

16
source/libs/sync/inc/syncEnv.h
View File

@ -20,13 +20,7 @@
extern "C" { extern "C" {
#endif #endif
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include "syncInt.h" #include "syncInt.h"
#include "taosdef.h"
#include "trpc.h"
#include "ttimer.h"
#define TIMER_MAX_MS 0x7FFFFFFF #define TIMER_MAX_MS 0x7FFFFFFF
#define ENV_TICK_TIMER_MS 1000 #define ENV_TICK_TIMER_MS 1000
@ -57,12 +51,12 @@ typedef struct SSyncEnv {
} SSyncEnv; } SSyncEnv;
extern SSyncEnv* gSyncEnv; SSyncEnv* syncEnv();
int32_t syncEnvStart(); int64_t syncNodeAdd(SSyncNode* pNode);
int32_t syncEnvStop(); void syncNodeRemove(SSyncNode* pNode);
int32_t syncEnvStartTimer(); SSyncNode* syncNodeAcquire(int64_t rid);
int32_t syncEnvStopTimer(); void syncNodeRelease(SSyncNode* pNode);
#ifdef __cplusplus #ifdef __cplusplus
} }

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

@ -24,6 +24,8 @@ extern "C" {
#include "syncTools.h" #include "syncTools.h"
#include "tlog.h" #include "tlog.h"
#include "ttimer.h" #include "ttimer.h"
#include "taosdef.h"
#include "ttimer.h"
// clang-format off // clang-format off
#define sFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }} while(0) #define sFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }} while(0)
@ -255,9 +257,6 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, Sync
SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode); SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode);
char* syncNodePeerState2Str(const SSyncNode* pSyncNode); char* syncNodePeerState2Str(const SSyncNode* pSyncNode);
SSyncNode* syncNodeAcquire(int64_t rid);
void syncNodeRelease(SSyncNode* pNode);
// raft state change -------------- // raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term); void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);
void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term); void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term);
@ -302,9 +301,6 @@ bool syncNodeNeedSendAppendEntries(SSyncNode* ths, const SRaftId* pDestId, const
int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta); int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta); int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta);
void syncStartNormal(int64_t rid);
void syncStartStandBy(int64_t rid);
bool syncNodeCanChange(SSyncNode* pSyncNode); bool syncNodeCanChange(SSyncNode* pSyncNode);
bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg); bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg);

184
source/libs/sync/src/syncEnv.c
View File

@ -13,118 +13,114 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _DEFAULT_SOURCE
#include "syncEnv.h" #include "syncEnv.h"
// #include <ASSERT.h> #include "tref.h"
SSyncEnv *gSyncEnv = NULL; static SSyncEnv gSyncEnv = {0};
static int32_t gNodeRefId = -1;
bool gRaftDetailLog = false;
static void syncEnvTick(void *param, void *tmrId);
// local function ----------------- SSyncEnv *syncEnv() { return &gSyncEnv; }
static SSyncEnv *doSyncEnvStart();
static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv);
static int32_t doSyncEnvStartTimer(SSyncEnv *pSyncEnv);
static int32_t doSyncEnvStopTimer(SSyncEnv *pSyncEnv);
static void syncEnvTick(void *param, void *tmrId);
// --------------------------------
bool syncEnvIsStart() { bool syncIsInit() { return atomic_load_8(&gSyncEnv.isStart); }
if (gSyncEnv == NULL) {
return false;
}
return atomic_load_8(&(gSyncEnv->isStart)); int32_t syncInit() {
} if (syncIsInit()) return 0;
int32_t syncEnvStart() {
int32_t ret = 0;
uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF); uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF);
taosSeedRand(seed); taosSeedRand(seed);
// gSyncEnv = doSyncEnvStart(gSyncEnv);
gSyncEnv = doSyncEnvStart();
ASSERT(gSyncEnv != NULL);
sTrace("sync env start ok");
return ret;
}
int32_t syncEnvStop() { memset(&gSyncEnv, 0, sizeof(SSyncEnv));
int32_t ret = doSyncEnvStop(gSyncEnv); gSyncEnv.envTickTimerCounter = 0;
return ret; gSyncEnv.envTickTimerMS = ENV_TICK_TIMER_MS;
} gSyncEnv.FpEnvTickTimer = syncEnvTick;
atomic_store_64(&gSyncEnv.envTickTimerLogicClock, 0);
int32_t syncEnvStartTimer() { atomic_store_64(&gSyncEnv.envTickTimerLogicClockUser, 0);
int32_t ret = doSyncEnvStartTimer(gSyncEnv);
return ret;
}
int32_t syncEnvStopTimer() {
int32_t ret = doSyncEnvStopTimer(gSyncEnv);
return ret;
}
// local function -----------------
static void syncEnvTick(void *param, void *tmrId) {
SSyncEnv *pSyncEnv = (SSyncEnv *)param;
if (atomic_load_64(&pSyncEnv->envTickTimerLogicClockUser) <= atomic_load_64(&pSyncEnv->envTickTimerLogicClock)) {
++(pSyncEnv->envTickTimerCounter);
sTrace("syncEnvTick do ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
", envTickTimerCounter:%" PRIu64
", "
"envTickTimerMS:%d, tmrId:%p",
pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
pSyncEnv->envTickTimerMS, tmrId);
// do something, tick ...
taosTmrReset(syncEnvTick, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager, &pSyncEnv->pEnvTickTimer);
} else {
sTrace("syncEnvTick pass ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
", envTickTimerCounter:%" PRIu64
", "
"envTickTimerMS:%d, tmrId:%p",
pSyncEnv->envTickTimerLogicClockUser, pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerCounter,
pSyncEnv->envTickTimerMS, tmrId);
}
}
static SSyncEnv *doSyncEnvStart() {
SSyncEnv *pSyncEnv = (SSyncEnv *)taosMemoryMalloc(sizeof(SSyncEnv));
ASSERT(pSyncEnv != NULL);
memset(pSyncEnv, 0, sizeof(SSyncEnv));
pSyncEnv->envTickTimerCounter = 0;
pSyncEnv->envTickTimerMS = ENV_TICK_TIMER_MS;
pSyncEnv->FpEnvTickTimer = syncEnvTick;
atomic_store_64(&pSyncEnv->envTickTimerLogicClock, 0);
atomic_store_64(&pSyncEnv->envTickTimerLogicClockUser, 0);
// start tmr thread // start tmr thread
pSyncEnv->pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV"); gSyncEnv.pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");
atomic_store_8(&gSyncEnv.isStart, 1);
atomic_store_8(&(pSyncEnv->isStart), 1); gNodeRefId = taosOpenRef(200, (RefFp)syncNodeClose);
return pSyncEnv; if (gNodeRefId < 0) {
} sError("failed to init node ref");
syncCleanUp();
static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) { return -1;
ASSERT(pSyncEnv == gSyncEnv);
if (pSyncEnv != NULL) {
atomic_store_8(&(pSyncEnv->isStart), 0);
taosTmrCleanUp(pSyncEnv->pTimerManager);
taosMemoryFree(pSyncEnv);
} }
gSyncEnv = NULL;
sDebug("sync rsetId:%d is open", gNodeRefId);
return 0; return 0;
} }
static int32_t doSyncEnvStartTimer(SSyncEnv *pSyncEnv) { void syncCleanUp() {
int32_t ret = 0; atomic_store_8(&gSyncEnv.isStart, 0);
taosTmrReset(pSyncEnv->FpEnvTickTimer, pSyncEnv->envTickTimerMS, pSyncEnv, pSyncEnv->pTimerManager, taosTmrCleanUp(gSyncEnv.pTimerManager);
&pSyncEnv->pEnvTickTimer); memset(&gSyncEnv, 0, sizeof(SSyncEnv));
atomic_store_64(&pSyncEnv->envTickTimerLogicClock, pSyncEnv->envTickTimerLogicClockUser);
return ret; if (gNodeRefId != -1) {
sDebug("sync rsetId:%d is closed", gNodeRefId);
taosCloseRef(gNodeRefId);
gNodeRefId = -1;
}
} }
static int32_t doSyncEnvStopTimer(SSyncEnv *pSyncEnv) { int64_t syncNodeAdd(SSyncNode *pNode) {
pNode->rid = taosAddRef(gNodeRefId, pNode);
if (pNode->rid < 0) return -1;
sDebug("vgId:%d, sync rid:%" PRId64 " is added to rsetId:%d", pNode->vgId, pNode->rid, gNodeRefId);
return 0;
}
void syncNodeRemove(SSyncNode *pNode) {
taosRemoveRef(gNodeRefId, pNode->rid);
sDebug("vgId:%d, sync rid:%" PRId64 " is removed from rsetId:%d", pNode->vgId, pNode->rid, gNodeRefId);
}
SSyncNode *syncNodeAcquire(int64_t rid) {
SSyncNode *pNode = taosAcquireRef(gNodeRefId, rid);
if (pNode == NULL) {
sTrace("failed to acquire node from refId:%" PRId64, rid);
}
return pNode;
}
void syncNodeRelease(SSyncNode *pNode) { taosReleaseRef(gNodeRefId, pNode->rid); }
#if 0
void syncEnvStartTimer() {
taosTmrReset(gSyncEnv.FpEnvTickTimer, gSyncEnv.envTickTimerMS, &gSyncEnv, gSyncEnv.pTimerManager,
&gSyncEnv.pEnvTickTimer);
atomic_store_64(&gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerLogicClockUser);
}
void syncEnvStopTimer() {
int32_t ret = 0; int32_t ret = 0;
atomic_add_fetch_64(&pSyncEnv->envTickTimerLogicClockUser, 1); atomic_add_fetch_64(&gSyncEnv.envTickTimerLogicClockUser, 1);
taosTmrStop(pSyncEnv->pEnvTickTimer); taosTmrStop(gSyncEnv.pEnvTickTimer);
pSyncEnv->pEnvTickTimer = NULL; gSyncEnv.pEnvTickTimer = NULL;
return ret; return ret;
} }
#endif
static void syncEnvTick(void *param, void *tmrId) {
SSyncEnv *pSyncEnv = param;
if (atomic_load_64(&gSyncEnv.envTickTimerLogicClockUser) <= atomic_load_64(&gSyncEnv.envTickTimerLogicClock)) {
gSyncEnv.envTickTimerCounter++;
sTrace("syncEnvTick do ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
", envTickTimerCounter:%" PRIu64 ", envTickTimerMS:%d, tmrId:%p",
gSyncEnv.envTickTimerLogicClockUser, gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerCounter,
gSyncEnv.envTickTimerMS, tmrId);
// do something, tick ...
taosTmrReset(syncEnvTick, gSyncEnv.envTickTimerMS, pSyncEnv, gSyncEnv.pTimerManager, &gSyncEnv.pEnvTickTimer);
} else {
sTrace("syncEnvTick pass ... envTickTimerLogicClockUser:%" PRIu64 ", envTickTimerLogicClock:%" PRIu64
", envTickTimerCounter:%" PRIu64 ", envTickTimerMS:%d, tmrId:%p",
gSyncEnv.envTickTimerLogicClockUser, gSyncEnv.envTickTimerLogicClock, gSyncEnv.envTickTimerCounter,
gSyncEnv.envTickTimerMS, tmrId);
}
}

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

@ -33,11 +33,6 @@
#include "syncTimeout.h" #include "syncTimeout.h"
#include "syncUtil.h" #include "syncUtil.h"
#include "syncVoteMgr.h" #include "syncVoteMgr.h"
#include "tref.h"
bool gRaftDetailLog = false;
static int32_t tsNodeRefId = -1;
// ------ local funciton --------- // ------ local funciton ---------
// enqueue message ---- // enqueue message ----
@ -52,138 +47,36 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId);
int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg); int32_t syncNodeOnPing(SSyncNode* ths, SyncPing* pMsg);
int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg); int32_t syncNodeOnPingReply(SSyncNode* ths, SyncPingReply* pMsg);
// ---------------------------------
static void syncNodeFreeCb(void* param) {
syncNodeClose(param);
param = NULL;
}
int32_t syncInit() {
int32_t ret = 0;
if (!syncEnvIsStart()) {
tsNodeRefId = taosOpenRef(200, syncNodeFreeCb);
if (tsNodeRefId < 0) {
sError("failed to init node ref");
syncCleanUp();
ret = -1;
} else {
sDebug("sync rsetId:%d is open", tsNodeRefId);
ret = syncEnvStart();
}
}
return ret;
}
void syncCleanUp() {
int32_t ret = syncEnvStop();
ASSERT(ret == 0);
if (tsNodeRefId != -1) {
sDebug("sync rsetId:%d is closed", tsNodeRefId);
taosCloseRef(tsNodeRefId);
tsNodeRefId = -1;
}
}
int64_t syncOpen(SSyncInfo* pSyncInfo) { int64_t syncOpen(SSyncInfo* pSyncInfo) {
SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo); SSyncNode* pNode = syncNodeOpen(pSyncInfo);
if (pSyncNode == NULL) { if (pNode == NULL) {
sError("vgId:%d, failed to open sync node", pSyncInfo->vgId); sError("vgId:%d, failed to open sync node", pSyncInfo->vgId);
return -1; return -1;
} }
pSyncNode->rid = taosAddRef(tsNodeRefId, pSyncNode); pNode->rid = syncNodeAdd(pNode);
if (pSyncNode->rid < 0) { if (pNode->rid < 0) {
syncNodeClose(pSyncNode); syncNodeClose(pNode);
pSyncNode = NULL;
return -1; return -1;
} }
sDebug("vgId:%d, sync rid:%" PRId64 " is added to rsetId:%d", pSyncInfo->vgId, pSyncNode->rid, tsNodeRefId); return pNode->rid;
return pSyncNode->rid;
} }
void syncStart(int64_t rid) { void syncStart(int64_t rid) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid); SSyncNode* pNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pNode != NULL) {
return; syncNodeStart(pNode);
syncNodeRelease(pNode);
} }
if (pSyncNode->pRaftCfg->isStandBy) {
syncNodeStartStandBy(pSyncNode);
} else {
syncNodeStart(pSyncNode);
}
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}
void syncStartNormal(int64_t rid) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
return;
}
syncNodeStart(pSyncNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}
void syncStartStandBy(int64_t rid) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
return;
}
syncNodeStartStandBy(pSyncNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
} }
void syncStop(int64_t rid) { void syncStop(int64_t rid) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid); SSyncNode* pNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) return; if (pNode != NULL) {
int32_t vgId = pSyncNode->vgId; syncNodeRelease(pNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRemove(pNode);
taosRemoveRef(tsNodeRefId, rid);
sDebug("vgId:%d, sync rid:%" PRId64 " is removed from rsetId:%d", vgId, rid, tsNodeRefId);
}
int32_t syncSetStandby(int64_t rid) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
sError("failed to set standby since accquire ref error, rid:%" PRId64, rid);
return -1;
} }
if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) {
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
terrno = TSDB_CODE_SYN_IS_LEADER;
} else {
terrno = TSDB_CODE_SYN_STANDBY_NOT_READY;
}
sError("failed to set standby since it is not follower, state:%s rid:%" PRId64, syncStr(pSyncNode->state), rid);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return -1;
}
// state change
pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
syncNodeStopHeartbeatTimer(pSyncNode);
// reset elect timer, long enough
int32_t electMS = TIMER_MAX_MS;
int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS);
ASSERT(ret == 0);
pSyncNode->pRaftCfg->isStandBy = 1;
raftCfgPersist(pSyncNode->pRaftCfg);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
sInfo("vgId:%d, set to standby", pSyncNode->vgId);
return 0;
} }
bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg) { bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg) {
@ -204,7 +97,7 @@ bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg) {
} }
int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) { int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -213,7 +106,7 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
int32_t ret = 0; int32_t ret = 0;
if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) { if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR; terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
sError("invalid new config. vgId:%d", pSyncNode->vgId); sError("invalid new config. vgId:%d", pSyncNode->vgId);
return -1; return -1;
@ -227,12 +120,12 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
snprintf(pRpcMsg->pCont, pRpcMsg->contLen, "%s", newconfig); snprintf(pRpcMsg->pCont, pRpcMsg->contLen, "%s", newconfig);
taosMemoryFree(newconfig); taosMemoryFree(newconfig);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) { int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -240,7 +133,7 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) { if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR; terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
sError("invalid new config. vgId:%d", pSyncNode->vgId); sError("invalid new config. vgId:%d", pSyncNode->vgId);
return -1; return -1;
@ -259,7 +152,7 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
taosMemoryFree(newconfig); taosMemoryFree(newconfig);
ret = syncNodePropose(pSyncNode, &rpcMsg, false); ret = syncNodePropose(pSyncNode, &rpcMsg, false);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
#else #else
syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg); syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg);
@ -275,13 +168,13 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
syncNodeReplicate(pSyncNode); syncNodeReplicate(pSyncNode);
} }
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
#endif #endif
} }
int32_t syncLeaderTransfer(int64_t rid) { int32_t syncLeaderTransfer(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -289,12 +182,12 @@ int32_t syncLeaderTransfer(int64_t rid) {
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
int32_t ret = syncNodeLeaderTransfer(pSyncNode); int32_t ret = syncNodeLeaderTransfer(pSyncNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) { int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -302,7 +195,7 @@ int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
int32_t ret = syncNodeLeaderTransferTo(pSyncNode, newLeader); int32_t ret = syncNodeLeaderTransferTo(pSyncNode, newLeader);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
@ -358,7 +251,7 @@ char* syncNodePeerState2Str(const SSyncNode* pSyncNode) {
} }
int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) { int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -382,7 +275,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
logNum, isEmpty); logNum, isEmpty);
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
@ -411,7 +304,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
} while (0); } while (0);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
} }
@ -424,7 +317,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
lastApplyIndex, pSyncNode->minMatchIndex); lastApplyIndex, pSyncNode->minMatchIndex);
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
@ -433,7 +326,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
snprintf(logBuf, sizeof(logBuf), "new-snapshot-index:%" PRId64 " candidate, do not delete wal", lastApplyIndex); snprintf(logBuf, sizeof(logBuf), "new-snapshot-index:%" PRId64 " candidate, do not delete wal", lastApplyIndex);
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} else { } else {
@ -442,7 +335,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
lastApplyIndex); lastApplyIndex);
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
@ -491,12 +384,12 @@ _DEL_WAL:
} }
} while (0); } while (0);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return code; return code;
} }
int32_t syncEndSnapshot(int64_t rid) { int32_t syncEndSnapshot(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -510,7 +403,7 @@ int32_t syncEndSnapshot(int64_t rid) {
if (code != 0) { if (code != 0) {
sError("vgId:%d, wal snapshot end error since:%s", pSyncNode->vgId, terrstr(terrno)); sError("vgId:%d, wal snapshot end error since:%s", pSyncNode->vgId, terrstr(terrno));
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return -1; return -1;
} else { } else {
do { do {
@ -524,12 +417,12 @@ int32_t syncEndSnapshot(int64_t rid) {
} }
} }
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return code; return code;
} }
int32_t syncStepDown(int64_t rid, SyncTerm newTerm) { int32_t syncStepDown(int64_t rid, SyncTerm newTerm) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
@ -538,7 +431,7 @@ int32_t syncStepDown(int64_t rid, SyncTerm newTerm) {
syncNodeStepDown(pSyncNode, newTerm); syncNodeStepDown(pSyncNode, newTerm);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
@ -583,19 +476,19 @@ int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) {
} }
bool syncCanLeaderTransfer(int64_t rid) { bool syncCanLeaderTransfer(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return false; return false;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
if (pSyncNode->replicaNum == 1) { if (pSyncNode->replicaNum == 1) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return false; return false;
} }
if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) { if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return true; return true;
} }
@ -610,7 +503,7 @@ bool syncCanLeaderTransfer(int64_t rid) {
} }
} }
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return matchOK; return matchOK;
} }
@ -620,25 +513,25 @@ int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
} }
ESyncState syncGetMyRole(int64_t rid) { ESyncState syncGetMyRole(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_STATE_ERROR; return TAOS_SYNC_STATE_ERROR;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
ESyncState state = pSyncNode->state; ESyncState state = pSyncNode->state;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return state; return state;
} }
bool syncIsReady(int64_t rid) { bool syncIsReady(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return false; return false;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && pSyncNode->restoreFinish; bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && pSyncNode->restoreFinish;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
// if false, set error code // if false, set error code
if (false == b) { if (false == b) {
@ -652,14 +545,14 @@ bool syncIsReady(int64_t rid) {
} }
bool syncIsRestoreFinish(int64_t rid) { bool syncIsRestoreFinish(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return false; return false;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
bool b = pSyncNode->restoreFinish; bool b = pSyncNode->restoreFinish;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return b; return b;
} }
@ -668,7 +561,7 @@ int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapsho
return -1; return -1;
} }
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return -1; return -1;
} }
@ -680,7 +573,7 @@ int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapsho
if (pEntry != NULL) { if (pEntry != NULL) {
syncEntryDestory(pEntry); syncEntryDestory(pEntry);
} }
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return -1; return -1;
} }
ASSERT(pEntry != NULL); ASSERT(pEntry != NULL);
@ -691,12 +584,12 @@ int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapsho
pSnapshot->lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, index); pSnapshot->lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, index);
syncEntryDestory(pEntry); syncEntryDestory(pEntry);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) { int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return -1; return -1;
} }
@ -705,12 +598,12 @@ int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) {
sTrace("vgId:%d, get snapshot meta, lastConfigIndex:%" PRId64, pSyncNode->vgId, pSyncNode->pRaftCfg->lastConfigIndex); sTrace("vgId:%d, get snapshot meta, lastConfigIndex:%" PRId64, pSyncNode->vgId, pSyncNode->pRaftCfg->lastConfigIndex);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) { int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return -1; return -1;
} }
@ -729,7 +622,7 @@ int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct
sTrace("vgId:%d, get snapshot meta by index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId, snapshotIndex, sTrace("vgId:%d, get snapshot meta by index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId, snapshotIndex,
sMeta->lastConfigIndex); sMeta->lastConfigIndex);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return 0; return 0;
} }
@ -755,67 +648,67 @@ const char* syncGetMyRoleStr(int64_t rid) {
} }
bool syncRestoreFinish(int64_t rid) { bool syncRestoreFinish(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return false; return false;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
bool restoreFinish = pSyncNode->restoreFinish; bool restoreFinish = pSyncNode->restoreFinish;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return restoreFinish; return restoreFinish;
} }
SyncTerm syncGetMyTerm(int64_t rid) { SyncTerm syncGetMyTerm(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_STATE_ERROR; return TAOS_SYNC_STATE_ERROR;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
SyncTerm term = pSyncNode->pRaftStore->currentTerm; SyncTerm term = pSyncNode->pRaftStore->currentTerm;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return term; return term;
} }
SyncIndex syncGetLastIndex(int64_t rid) { SyncIndex syncGetLastIndex(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return SYNC_INDEX_INVALID; return SYNC_INDEX_INVALID;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return lastIndex; return lastIndex;
} }
SyncIndex syncGetCommitIndex(int64_t rid) { SyncIndex syncGetCommitIndex(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return SYNC_INDEX_INVALID; return SYNC_INDEX_INVALID;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
SyncIndex cmtIndex = pSyncNode->commitIndex; SyncIndex cmtIndex = pSyncNode->commitIndex;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return cmtIndex; return cmtIndex;
} }
SyncGroupId syncGetVgId(int64_t rid) { SyncGroupId syncGetVgId(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_STATE_ERROR; return TAOS_SYNC_STATE_ERROR;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
SyncGroupId vgId = pSyncNode->vgId; SyncGroupId vgId = pSyncNode->vgId;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return vgId; return vgId;
} }
void syncGetEpSet(int64_t rid, SEpSet* pEpSet) { void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
memset(pEpSet, 0, sizeof(*pEpSet)); memset(pEpSet, 0, sizeof(*pEpSet));
return; return;
@ -831,11 +724,11 @@ void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex; pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex;
sInfo("vgId:%d, sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); sInfo("vgId:%d, sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) { void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
memset(pEpSet, 0, sizeof(*pEpSet)); memset(pEpSet, 0, sizeof(*pEpSet));
return; return;
@ -854,11 +747,11 @@ void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) {
} }
sInfo("vgId:%d, sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); sInfo("vgId:%d, sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg) { int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_STATE_ERROR; return TAOS_SYNC_STATE_ERROR;
} }
@ -870,12 +763,12 @@ int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg) {
memcpy(msg, &(stub.rpcMsg), sizeof(SRpcMsg)); memcpy(msg, &(stub.rpcMsg), sizeof(SRpcMsg));
} }
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo) { int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_STATE_ERROR; return TAOS_SYNC_STATE_ERROR;
} }
@ -888,12 +781,12 @@ int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo)
} }
sTrace("vgId:%d, get seq:%" PRIu64 " rpc handle:%p", pSyncNode->vgId, index, pInfo->handle); sTrace("vgId:%d, get seq:%" PRIu64 " rpc handle:%p", pSyncNode->vgId, index, pInfo->handle);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb) { void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
sTrace("syncSetQ get pSyncNode is NULL, rid:%" PRId64, rid); sTrace("syncSetQ get pSyncNode is NULL, rid:%" PRId64, rid);
return; return;
@ -901,24 +794,24 @@ void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb) {
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
pSyncNode->msgcb = msgcb; pSyncNode->msgcb = msgcb;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
char* sync2SimpleStr(int64_t rid) { char* sync2SimpleStr(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
sTrace("syncSetRpc get pSyncNode is NULL, rid:%" PRId64, rid); sTrace("syncSetRpc get pSyncNode is NULL, rid:%" PRId64, rid);
return NULL; return NULL;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
char* s = syncNode2SimpleStr(pSyncNode); char* s = syncNode2SimpleStr(pSyncNode);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return s; return s;
} }
void setPingTimerMS(int64_t rid, int32_t pingTimerMS) { void setPingTimerMS(int64_t rid, int32_t pingTimerMS) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return; return;
} }
@ -926,22 +819,22 @@ void setPingTimerMS(int64_t rid, int32_t pingTimerMS) {
pSyncNode->pingBaseLine = pingTimerMS; pSyncNode->pingBaseLine = pingTimerMS;
pSyncNode->pingTimerMS = pingTimerMS; pSyncNode->pingTimerMS = pingTimerMS;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
void setElectTimerMS(int64_t rid, int32_t electTimerMS) { void setElectTimerMS(int64_t rid, int32_t electTimerMS) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return; return;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
pSyncNode->electBaseLine = electTimerMS; pSyncNode->electBaseLine = electTimerMS;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) { void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return; return;
} }
@ -949,20 +842,20 @@ void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
pSyncNode->hbBaseLine = hbTimerMS; pSyncNode->hbBaseLine = hbTimerMS;
pSyncNode->heartbeatTimerMS = hbTimerMS; pSyncNode->heartbeatTimerMS = hbTimerMS;
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
} }
int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) { int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = syncNodeAcquire(rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
taosReleaseRef(tsNodeRefId, rid); syncNodeRelease(pSyncNode);
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1; return -1;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
int32_t ret = syncNodePropose(pSyncNode, pMsg, isWeak); int32_t ret = syncNodePropose(pSyncNode, pMsg, isWeak);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); syncNodeRelease(pSyncNode);
return ret; return ret;
} }
@ -1084,7 +977,7 @@ int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRaftId de
int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
int32_t ret = 0; int32_t ret = 0;
if (syncEnvIsStart()) { if (syncIsInit()) {
SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData)); SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData));
pData->pSyncNode = pSyncNode; pData->pSyncNode = pSyncNode;
pData->pTimer = pSyncTimer; pData->pTimer = pSyncTimer;
@ -1092,7 +985,7 @@ int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
pData->logicClock = pSyncTimer->logicClock; pData->logicClock = pSyncTimer->logicClock;
pSyncTimer->pData = pData; pSyncTimer->pData = pData;
taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, gSyncEnv->pTimerManager, &pSyncTimer->pTimer); taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager, &pSyncTimer->pTimer);
} else { } else {
sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
} }
@ -1556,8 +1449,8 @@ int32_t syncNodePingAll(SSyncNode* pSyncNode) {
// timer control -------------- // timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t ret = 0;
if (syncEnvIsStart()) { if (syncIsInit()) {
taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pPingTimer); &pSyncNode->pPingTimer);
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
} else { } else {
@ -1576,7 +1469,7 @@ int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
int32_t ret = 0; int32_t ret = 0;
if (syncEnvIsStart()) { if (syncIsInit()) {
pSyncNode->electTimerMS = ms; pSyncNode->electTimerMS = ms;
SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer)); SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer));
@ -1584,7 +1477,7 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
pElectTimer->pSyncNode = pSyncNode; pElectTimer->pSyncNode = pSyncNode;
pElectTimer->pData = NULL; pElectTimer->pData = NULL;
taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, syncEnv()->pTimerManager,
&pSyncNode->pElectTimer); &pSyncNode->pElectTimer);
} else { } else {
@ -1632,8 +1525,8 @@ int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode) {
static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) { static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t ret = 0;
if (syncEnvIsStart()) { if (syncIsInit()) {
taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser); atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} else { } else {
@ -2325,17 +2218,6 @@ _END:
return; return;
} }
SSyncNode* syncNodeAcquire(int64_t rid) {
SSyncNode* pNode = taosAcquireRef(tsNodeRefId, rid);
if (pNode == NULL) {
sTrace("failed to acquire node from refId:%" PRId64, rid);
}
return pNode;
}
void syncNodeRelease(SSyncNode* pNode) { taosReleaseRef(tsNodeRefId, pNode->rid); }
// raft state change -------------- // raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) { void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
if (term > pSyncNode->pRaftStore->currentTerm) { if (term > pSyncNode->pRaftStore->currentTerm) {
@ -2786,8 +2668,8 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
if (syncEnvIsStart()) { if (syncIsInit()) {
taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pPingTimer); &pSyncNode->pPingTimer);
} else { } else {
sError("sync env is stop, syncNodeEqPingTimer"); sError("sync env is stop, syncNodeEqPingTimer");
@ -2832,9 +2714,9 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
#if 0 #if 0
// reset timer ms // reset timer ms
if (syncEnvIsStart() && pSyncNode->electBaseLine > 0) { if (syncIsInit() && pSyncNode->electBaseLine > 0) {
pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine); pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pElectTimer); &pSyncNode->pElectTimer);
} else { } else {
sError("sync env is stop, syncNodeEqElectTimer"); sError("sync env is stop, syncNodeEqElectTimer");
@ -2869,8 +2751,8 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
if (syncEnvIsStart()) { if (syncIsInit()) {
taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
} else { } else {
sError("sync env is stop, syncNodeEqHeartbeatTimer"); sError("sync env is stop, syncNodeEqHeartbeatTimer");
@ -2930,8 +2812,8 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
syncHeartbeatDestroy(pSyncMsg); syncHeartbeatDestroy(pSyncMsg);
if (syncEnvIsStart()) { if (syncIsInit()) {
taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS, pData, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager,
&pSyncTimer->pTimer); &pSyncTimer->pTimer);
} else { } else {
sError("sync env is stop, syncNodeEqHeartbeatTimer"); sError("sync env is stop, syncNodeEqHeartbeatTimer");

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

@ -98,7 +98,7 @@ int main(int argc, char** argv) {
init(); init();
int32_t ret = syncIOStart((char*)"127.0.0.1", gPorts[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", gPorts[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
char walPath[128]; char walPath[128];

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

@ -152,7 +152,7 @@ int main(int argc, char **argv) {
int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
taosRemoveDir("./wal_test"); taosRemoveDir("./wal_test");

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

@ -81,7 +81,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

6
source/libs/sync/test/syncEnvTest.cpp
View File

@ -22,7 +22,7 @@ int main() {
logTest(); logTest();
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 5; ++i) {
@ -37,8 +37,6 @@ int main() {
taosMsleep(5000); taosMsleep(5000);
} }
ret = syncEnvStop(); syncCleanUp();
assert(ret == 0);
return 0; return 0;
} }

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

@ -82,7 +82,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -81,7 +81,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();
@ -91,7 +91,7 @@ int main(int argc, char** argv) {
initRaftId(pSyncNode); initRaftId(pSyncNode);
syncNodeClose(pSyncNode); syncNodeClose(pSyncNode);
syncEnvStop(); syncCleanUp();
// syncIOStop(); // syncIOStop();
// taosCloseLog(); // taosCloseLog();

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

@ -81,7 +81,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -81,7 +81,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -81,7 +81,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -179,7 +179,7 @@ int main(int argc, char **argv) {
int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
// taosRemoveDir(pWalDir); // taosRemoveDir(pWalDir);

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

@ -82,7 +82,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -82,7 +82,7 @@ int main(int argc, char** argv) {
int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
SSyncNode* pSyncNode = syncInitTest(); SSyncNode* pSyncNode = syncInitTest();

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

@ -154,7 +154,7 @@ int main(int argc, char **argv) {
int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
assert(ret == 0); assert(ret == 0);
ret = syncEnvStart(); ret = syncInit();
assert(ret == 0); assert(ret == 0);
taosRemoveDir("./wal_test"); taosRemoveDir("./wal_test");

2
source/util/src/tref.c
View File

@ -57,7 +57,7 @@ static void taosIncRsetCount(SRefSet *pSet);
static void taosDecRsetCount(SRefSet *pSet); static void taosDecRsetCount(SRefSet *pSet);
static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove); static int32_t taosDecRefCount(int32_t rsetId, int64_t rid, int32_t remove);
int32_t taosOpenRef(int32_t max, void (*fp)(void *)) { int32_t taosOpenRef(int32_t max, RefFp fp) {
SRefNode **nodeList; SRefNode **nodeList;
SRefSet *pSet; SRefSet *pSet;
int64_t *lockedBy; int64_t *lockedBy;