From 82144e35f0e8ed69f75acf931c98252b16c25402 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 16:00:57 +0800 Subject: [PATCH 1/6] sync refactor --- source/libs/sync/test/CMakeLists.txt | 14 ++ source/libs/sync/test/syncElectTest.cpp | 7 +- source/libs/sync/test/syncRefTest.cpp | 134 ++++++++++++++++++++ source/libs/sync/test/syncReplicateTest.cpp | 16 ++- 4 files changed, 162 insertions(+), 9 deletions(-) create mode 100644 source/libs/sync/test/syncRefTest.cpp diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 0c7608f902..3fcb9087ab 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -31,6 +31,7 @@ add_executable(syncElectTest "") add_executable(syncEncodeTest "") add_executable(syncWriteTest "") add_executable(syncReplicateTest "") +add_executable(syncRefTest "") target_sources(syncTest @@ -165,6 +166,10 @@ target_sources(syncReplicateTest PRIVATE "syncReplicateTest.cpp" ) +target_sources(syncRefTest + PRIVATE + "syncRefTest.cpp" +) target_include_directories(syncTest @@ -337,6 +342,11 @@ target_include_directories(syncReplicateTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncRefTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -471,6 +481,10 @@ target_link_libraries(syncReplicateTest sync gtest_main ) +target_link_libraries(syncRefTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncElectTest.cpp b/source/libs/sync/test/syncElectTest.cpp index 0e0b57a025..47f55aa660 100644 --- a/source/libs/sync/test/syncElectTest.cpp +++ b/source/libs/sync/test/syncElectTest.cpp @@ -116,9 +116,10 @@ int main(int argc, char** argv) { //--------------------------- while (1) { - sTrace("elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", - gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, - gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); + sTrace( + "elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, + gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); } return 0; diff --git a/source/libs/sync/test/syncRefTest.cpp b/source/libs/sync/test/syncRefTest.cpp new file mode 100644 index 0000000000..8e6061a8ae --- /dev/null +++ b/source/libs/sync/test/syncRefTest.cpp @@ -0,0 +1,134 @@ +#include +#include +#include "syncIO.h" +#include "syncInt.h" +#include "syncRaftStore.h" +#include "tref.h" + +void logTest() { + sTrace("--- sync log test: trace"); + sDebug("--- sync log test: debug"); + sInfo("--- sync log test: info"); + sWarn("--- sync log test: warn"); + sError("--- sync log test: error"); + sFatal("--- sync log test: fatal"); +} + +static void syncFreeObj(void *param); +int32_t init(); +void cleanup(); +int64_t start(); +void stop(int64_t rid); + +static int32_t tsNodeRefId = -1; +int g = 100; + +typedef struct SyncObj { + int64_t rid; + void * data; + char name[32]; + int counter; +} SyncObj; + +static void syncFreeObj(void *param) { + SyncObj *pObj = (SyncObj *)param; + printf("syncFreeObj name:%s rid:%ld \n", pObj->name, pObj->rid); + free(pObj); +} + +int32_t init() { + tsNodeRefId = taosOpenRef(200, syncFreeObj); + if (tsNodeRefId < 0) { + sError("failed to init node ref"); + cleanup(); + return -1; + } + return 0; +} + +void cleanup() { + if (tsNodeRefId != -1) { + taosCloseRef(tsNodeRefId); + tsNodeRefId = -1; + } +} + +int64_t start() { + SyncObj *pObj = (SyncObj *)malloc(sizeof(SyncObj)); + assert(pObj != NULL); + + pObj->data = &g; + snprintf(pObj->name, sizeof(pObj->name), "%s", "hello"); + + pObj->rid = taosAddRef(tsNodeRefId, pObj); + if (pObj->rid < 0) { + syncFreeObj(pObj); + return -1; + } + + printf("start name:%s rid:%ld \n", pObj->name, pObj->rid); + return pObj->rid; +} + +void stop(int64_t rid) { + SyncObj *pObj = (SyncObj *)taosAcquireRef(tsNodeRefId, rid); + if (pObj == NULL) return; + + printf("stop name:%s rid:%ld \n", pObj->name, pObj->rid); + pObj->data = NULL; + + taosReleaseRef(tsNodeRefId, pObj->rid); + taosRemoveRef(tsNodeRefId, rid); +} + +void *func(void *param) { + int64_t rid = (int64_t)param; + + int32_t ms = taosRand() % 10000; + taosMsleep(ms); + + SyncObj *pObj = (SyncObj *)taosAcquireRef(tsNodeRefId, rid); + if (pObj != NULL) { + printf("taosAcquireRef sleep:%d, name:%s, rid:%ld \n", ms, pObj->name, pObj->rid); + } else { + printf("taosAcquireRef sleep:%d, NULL! \n", ms); + } + + taosReleaseRef(tsNodeRefId, rid); +} + +int main() { + // taosInitLog((char *)"syncTest.log", 100000, 10); + tsAsyncLog = 0; + sDebugFlag = 143 + 64; + logTest(); + + taosSeedRand(taosGetTimestampSec()); + + int32_t ret; + + ret = init(); + assert(ret == 0); + + int64_t rid = start(); + assert(rid > 0); + + for (int i = 0; i < 20; ++i) { + pthread_t tid; + pthread_create(&tid, NULL, func, (void*)rid); + } + + int32_t ms = taosRand() % 10000; + taosMsleep(ms); + printf("main sleep %d, stop and clean ", ms); + + stop(rid); + cleanup(); + + while (1) { + taosMsleep(1000); + printf("sleep 1 ... \n"); + } + + return 0; +} diff --git a/source/libs/sync/test/syncReplicateTest.cpp b/source/libs/sync/test/syncReplicateTest.cpp index 6c4fab2425..4d6e6f3a25 100644 --- a/source/libs/sync/test/syncReplicateTest.cpp +++ b/source/libs/sync/test/syncReplicateTest.cpp @@ -172,15 +172,19 @@ int main(int argc, char **argv) { gSyncNode->FpEqMsg(gSyncNode->queue, &rpcMsg); taosMsleep(1000); - sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", - gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, - gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); + sTrace( + "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " + "electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, + gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); } while (1) { - sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", - gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, - gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); + sTrace( + "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " + "electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, + gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); taosMsleep(1000); } From 7129645023bb798bb617f00a0215c8598997239c Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 16:28:50 +0800 Subject: [PATCH 2/6] sync refactor --- source/libs/sync/inc/syncInt.h | 1 - source/libs/sync/src/syncMain.c | 52 +++++++++++++++++++++++---- source/libs/sync/test/syncRefTest.cpp | 4 +-- 3 files changed, 47 insertions(+), 10 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 8e36424f19..ee009d6428 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -142,7 +142,6 @@ typedef struct SSyncNode { SRaftId leaderCache; // life cycle - int32_t refCount; int64_t rid; // tla+ server vars diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 829526512b..4b3283fc53 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -31,6 +31,7 @@ #include "syncTimeout.h" #include "syncUtil.h" #include "syncVoteMgr.h" +#include "tref.h" static int32_t tsNodeRefId = -1; @@ -44,31 +45,57 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId); static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg); static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg); static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg); + +// life cycle +static void syncFreeNode(void* param); // --------------------------------- int32_t syncInit() { - int32_t ret = syncEnvStart(); + int32_t ret; + tsNodeRefId = taosOpenRef(200, syncFreeNode); + if (tsNodeRefId < 0) { + sError("failed to init node ref"); + syncCleanUp(); + ret = -1; + } else { + ret = syncEnvStart(); + } + return ret; } void syncCleanUp() { int32_t ret = syncEnvStop(); assert(ret == 0); + + if (tsNodeRefId != -1) { + taosCloseRef(tsNodeRefId); + tsNodeRefId = -1; + } } int64_t syncStart(const SSyncInfo* pSyncInfo) { - int32_t ret = 0; SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo); assert(pSyncNode != NULL); - // todo : return ref id - return ret; + pSyncNode->rid = taosAddRef(tsNodeRefId, pSyncNode); + if (pSyncNode->rid < 0) { + syncFreeNode(pSyncNode); + return -1; + } + + return pSyncNode->rid; } void syncStop(int64_t rid) { - // todo : get pointer from rid - SSyncNode* pSyncNode = NULL; + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return; + } syncNodeClose(pSyncNode); + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + taosRemoveRef(tsNodeRefId, rid); } int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) { @@ -155,7 +182,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) { pSyncNode->quorum = syncUtilQuorum(pSyncInfo->syncCfg.replicaNum); pSyncNode->leaderCache = EMPTY_RAFT_ID; - // init life cycle + // init life cycle outside // TLA+ Spec // InitHistoryVars == /\ elections = {} @@ -444,6 +471,10 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) { cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache); cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache); + // life cycle + snprintf(u64buf, sizeof(u64buf), "%ld", pSyncNode->rid); + cJSON_AddStringToObject(pRoot, "rid", u64buf); + // tla+ server vars cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state); cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state)); @@ -813,3 +844,10 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg syncEntryDestory(pEntry); return ret; } + +static void syncFreeNode(void* param) { + SSyncNode* pNode = param; + syncNodePrint2((char*)"==syncFreeNode==", pNode); + + free(pNode); +} \ No newline at end of file diff --git a/source/libs/sync/test/syncRefTest.cpp b/source/libs/sync/test/syncRefTest.cpp index 8e6061a8ae..e5606af14f 100644 --- a/source/libs/sync/test/syncRefTest.cpp +++ b/source/libs/sync/test/syncRefTest.cpp @@ -82,7 +82,7 @@ void stop(int64_t rid) { } void *func(void *param) { - int64_t rid = (int64_t)param; + int64_t rid = (int64_t)param; int32_t ms = taosRand() % 10000; taosMsleep(ms); @@ -115,7 +115,7 @@ int main() { for (int i = 0; i < 20; ++i) { pthread_t tid; - pthread_create(&tid, NULL, func, (void*)rid); + pthread_create(&tid, NULL, func, (void *)rid); } int32_t ms = taosRand() % 10000; From 1251d64e8175d109cbf63d67ffab1290941eb4dc Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 16:42:30 +0800 Subject: [PATCH 3/6] sync refactor --- source/libs/sync/src/syncMain.c | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 4b3283fc53..c4adb4762b 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -103,11 +103,16 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) { return ret; } -int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { +int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { int32_t ret = 0; // todo : get pointer from rid - SSyncNode* pSyncNode = NULL; + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return -1; + } + assert(rid == pSyncNode->rid); + if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, 0, isWeak); SRpcMsg rpcMsg; @@ -120,6 +125,13 @@ int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { sTrace("syncForwardToPeer not leader, %s", syncUtilState2String(pSyncNode->state)); ret = -1; // todo : need define err code !! } + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return ret; +} + +int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { + int32_t ret = syncPropose(rid, pMsg, isWeak); return ret; } From 79e8f8e6ea6cb4be3dda1444516217a1b78cfb3b Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 16:50:13 +0800 Subject: [PATCH 4/6] sync refactor --- include/libs/sync/sync.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 68ca9eff17..a38431a1b2 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -157,7 +157,8 @@ void syncCleanUp(); int64_t syncStart(const SSyncInfo* pSyncInfo); void syncStop(int64_t rid); int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg); -int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak); +int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak); // use this function +int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak); // just for compatibility ESyncState syncGetMyRole(int64_t rid); void syncGetNodesRole(int64_t rid, SNodesRole* pNodeRole); From b8a13daf9f808b2cb83f889f58cc6602c04b98f4 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 16:56:37 +0800 Subject: [PATCH 5/6] sync refactor --- source/libs/sync/test/syncRefTest.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/sync/test/syncRefTest.cpp b/source/libs/sync/test/syncRefTest.cpp index e5606af14f..1a75ffa75f 100644 --- a/source/libs/sync/test/syncRefTest.cpp +++ b/source/libs/sync/test/syncRefTest.cpp @@ -114,8 +114,8 @@ int main() { assert(rid > 0); for (int i = 0; i < 20; ++i) { - pthread_t tid; - pthread_create(&tid, NULL, func, (void *)rid); + TdThread tid; + taosThreadCreate(&tid, NULL, func, (void *)rid); } int32_t ms = taosRand() % 10000; From 3a053453c69f6c30d812f99c32c84561e9caac0e Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Mon, 21 Mar 2022 17:07:00 +0800 Subject: [PATCH 6/6] sync refactor --- source/libs/sync/test/syncRefTest.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/source/libs/sync/test/syncRefTest.cpp b/source/libs/sync/test/syncRefTest.cpp index 1a75ffa75f..8e9adbbc15 100644 --- a/source/libs/sync/test/syncRefTest.cpp +++ b/source/libs/sync/test/syncRefTest.cpp @@ -95,6 +95,7 @@ void *func(void *param) { } taosReleaseRef(tsNodeRefId, rid); + return NULL; } int main() {