From 8057e44d1702af275db0cf30d085692c766b7fa5 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 2 Mar 2022 17:40:22 +0800 Subject: [PATCH] sync modify timer --- source/libs/sync/src/syncIO.c | 4 +++- source/libs/sync/src/syncMain.c | 14 +++++++++----- source/libs/sync/src/syncMessage.c | 4 ++-- source/libs/sync/src/syncUtil.c | 5 ++--- source/libs/sync/test/syncIOSendMsgClientTest.cpp | 2 +- source/libs/sync/test/syncIOSendMsgTest.cpp | 2 +- source/libs/sync/test/syncPingTest.cpp | 9 ++++++--- 7 files changed, 24 insertions(+), 16 deletions(-) diff --git a/source/libs/sync/src/syncIO.c b/source/libs/sync/src/syncIO.c index 88fba6429d..74f848338f 100644 --- a/source/libs/sync/src/syncIO.c +++ b/source/libs/sync/src/syncIO.c @@ -41,7 +41,8 @@ static void syncIOTickPingFunc(void *param, void *tmrId); // public function ------------ int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) { - sTrace("syncIOSendMsg ... "); + sTrace("<--- syncIOSendMsg ---> clientRpc:%p, numOfEps:%d, inUse:%d, destAddr:%s-%u", clientRpc, pEpSet->numOfEps, + pEpSet->inUse, pEpSet->eps[0].fqdn, pEpSet->eps[0].port); pMsg->handle = NULL; rpcSendRequest(clientRpc, pEpSet, pMsg, NULL); return 0; @@ -54,6 +55,7 @@ int32_t syncIOStart(char *host, uint16_t port) { int32_t ret = syncIOStartInternal(gSyncIO); assert(ret == 0); + sTrace("syncIOStart ok, gSyncIO:%p gSyncIO->clientRpc:%p", gSyncIO, gSyncIO->clientRpc); return 0; } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 9e335c5e55..d19c51900f 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -184,7 +184,7 @@ static int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, Syn SyncPing* pMsg2 = rpcMsg.pCont; cJSON* pJson = syncPing2Json(pMsg2); char* serialized = cJSON_Print(pJson); - sTrace("syncNodePing pMsg2:%s ", serialized); + sTrace("syncNodePing rpcMsg.pCont:%s ", serialized); free(serialized); cJSON_Delete(pJson); } @@ -253,12 +253,16 @@ static void syncNodePingTimerCb(void* param, void* tmrId) { ++(pSyncNode->pingTimerCounter); // pSyncNode->pingTimerMS += 100; - sTrace("pSyncNode->pingTimerCounter:%lu, pSyncNode->pingTimerMS:%d, pSyncNode->pPingTimer:%p, tmrId:%p ", - pSyncNode->pingTimerCounter, pSyncNode->pingTimerMS, pSyncNode->pPingTimer, tmrId); + sTrace( + "syncNodePingTimerCb: pSyncNode->pingTimerCounter:%lu, pSyncNode->pingTimerMS:%d, pSyncNode->pPingTimer:%p, " + "tmrId:%p ", + pSyncNode->pingTimerCounter, pSyncNode->pingTimerMS, pSyncNode->pPingTimer, tmrId); + + syncNodePingAll(pSyncNode); taosTmrReset(syncNodePingTimerCb, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager, &pSyncNode->pPingTimer); - - syncNodePingAll(pSyncNode); + } else { + sTrace("syncNodePingTimerCb: pingTimerStart:%u ", pSyncNode->pingTimerStart); } } \ No newline at end of file diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index e0975fb3c1..8e7dc8ba61 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -70,7 +70,7 @@ cJSON* syncPing2Json(const SyncPing* pMsg) { cJSON* pDestId = cJSON_CreateObject(); cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr); cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId); - cJSON_AddItemToObject(pRoot, "srcId", pDestId); + cJSON_AddItemToObject(pRoot, "destId", pDestId); cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen); cJSON_AddStringToObject(pRoot, "data", pMsg->data); @@ -145,7 +145,7 @@ cJSON* syncPingReply2Json(const SyncPingReply* pMsg) { cJSON* pDestId = cJSON_CreateObject(); cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr); cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId); - cJSON_AddItemToObject(pRoot, "srcId", pDestId); + cJSON_AddItemToObject(pRoot, "destId", pDestId); cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen); cJSON_AddStringToObject(pRoot, "data", pMsg->data); diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index 9c134e2d08..a3fbfaf905 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -51,11 +51,10 @@ void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet) { pEpSet->numOfEps = 1; pEpSet->inUse = 0; pEpSet->eps[0].port = port; - snprintf(epSet.eps[0].fqdn, sizeof(epSet.eps[0].fqdn), host); + snprintf(pEpSet->eps[0].fqdn, sizeof(pEpSet->eps[0].fqdn), "%s", host); */ - pEpSet->inUse = 0; - pEpSet->numOfEps = 1; + pEpSet->numOfEps = 0; addEpIntoEpSet(pEpSet, host, port); } diff --git a/source/libs/sync/test/syncIOSendMsgClientTest.cpp b/source/libs/sync/test/syncIOSendMsgClientTest.cpp index 09dc9bbb42..8adaff1a2f 100644 --- a/source/libs/sync/test/syncIOSendMsgClientTest.cpp +++ b/source/libs/sync/test/syncIOSendMsgClientTest.cpp @@ -34,7 +34,7 @@ int main() { SRpcMsg rpcMsg; rpcMsg.contLen = 64; rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); - snprintf((char*)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest"); + snprintf((char *)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest"); rpcMsg.handle = NULL; rpcMsg.msgType = 77; diff --git a/source/libs/sync/test/syncIOSendMsgTest.cpp b/source/libs/sync/test/syncIOSendMsgTest.cpp index 70fbe55b67..33af51f23d 100644 --- a/source/libs/sync/test/syncIOSendMsgTest.cpp +++ b/source/libs/sync/test/syncIOSendMsgTest.cpp @@ -34,7 +34,7 @@ int main() { SRpcMsg rpcMsg; rpcMsg.contLen = 64; rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); - snprintf((char*)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest"); + snprintf((char *)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest"); rpcMsg.handle = NULL; rpcMsg.msgType = 77; diff --git a/source/libs/sync/test/syncPingTest.cpp b/source/libs/sync/test/syncPingTest.cpp index 01e5731dde..902513f7cb 100644 --- a/source/libs/sync/test/syncPingTest.cpp +++ b/source/libs/sync/test/syncPingTest.cpp @@ -28,13 +28,16 @@ SSyncNode* doSync() { pCfg->replicaNum = 1; pCfg->nodeInfo[0].nodePort = 7010; - taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); + snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].nodeFqdn), "%s", "127.0.0.1"); + // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); pCfg->nodeInfo[1].nodePort = 7110; - taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn); + snprintf(pCfg->nodeInfo[1].nodeFqdn, sizeof(pCfg->nodeInfo[1].nodeFqdn), "%s", "127.0.0.1"); + // taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn); pCfg->nodeInfo[2].nodePort = 7210; - taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn); + snprintf(pCfg->nodeInfo[2].nodeFqdn, sizeof(pCfg->nodeInfo[2].nodeFqdn), "%s", "127.0.0.1"); + // taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn); SSyncNode* pSyncNode = syncNodeOpen(&syncInfo); assert(pSyncNode != NULL);