diff --git a/source/libs/sync/inc/syncRespMgr.h b/source/libs/sync/inc/syncRespMgr.h index 22e1005e5c495c6f4f17c452418d8b028d0cfffa..9026ecb66efc02f4e0e01d392063c1bdf4a5bdcb 100644 --- a/source/libs/sync/inc/syncRespMgr.h +++ b/source/libs/sync/inc/syncRespMgr.h @@ -41,13 +41,12 @@ typedef struct SSyncRespMgr { SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl); void syncRespMgrDestroy(SSyncRespMgr *pObj); -int64_t syncRespMgrAdd(SSyncRespMgr *pObj, SRespStub *pStub); -int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t index); -int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub); -int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub); +uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub); +int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq); +int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub); +int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo); void syncRespClean(SSyncRespMgr *pObj); void syncRespCleanRsp(SSyncRespMgr *pObj); -void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp); #ifdef __cplusplus } diff --git a/source/libs/sync/inc/syncTools.h b/source/libs/sync/inc/syncTools.h index 932432d1f89a78a2e86cc63b7c1ad94dc807f6a1..193579030f2c65b89bb74d21a4b9e87f38beb9dc 100644 --- a/source/libs/sync/inc/syncTools.h +++ b/source/libs/sync/inc/syncTools.h @@ -26,9 +26,6 @@ typedef struct SRaftId { SyncGroupId vgId; } SRaftId; -// for compatibility, the same as syncPropose -int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak); - // ------------------ for debug ------------------- void syncRpcMsgPrint(SRpcMsg* pMsg); void syncRpcMsgPrint2(char* s, SRpcMsg* pMsg); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 19015c00b9fae1e2c1adff68b54c80f5b65ad1f8..14823be0982be118d51163d52e5ec961fcdb7733 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -452,11 +452,6 @@ int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { return ret; } -int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak) { - int32_t ret = syncPropose(rid, pMsg, isWeak); - return ret; -} - SSyncState syncGetState(int64_t rid) { SSyncState state = {.state = TAOS_SYNC_STATE_ERROR}; @@ -558,109 +553,27 @@ SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapsho return lastIndex; } -#if 0 -SyncTerm syncGetMyTerm(int64_t rid) { - SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - return TAOS_SYNC_STATE_ERROR; - } - ASSERT(rid == pSyncNode->rid); - SyncTerm term = pSyncNode->pRaftStore->currentTerm; - - syncNodeRelease(pSyncNode); - return term; -} - -SyncIndex syncGetLastIndex(int64_t rid) { - SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - return SYNC_INDEX_INVALID; - } - ASSERT(rid == pSyncNode->rid); - SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode); - - syncNodeRelease(pSyncNode); - return lastIndex; -} - -SyncIndex syncGetCommitIndex(int64_t rid) { - SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - return SYNC_INDEX_INVALID; - } - ASSERT(rid == pSyncNode->rid); - SyncIndex cmtIndex = pSyncNode->commitIndex; - - syncNodeRelease(pSyncNode); - return cmtIndex; -} - -SyncGroupId syncGetVgId(int64_t rid) { - SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - return TAOS_SYNC_STATE_ERROR; - } - ASSERT(rid == pSyncNode->rid); - SyncGroupId vgId = pSyncNode->vgId; - - syncNodeRelease(pSyncNode); - return vgId; -} - -void syncGetEpSet(int64_t rid, SEpSet* pEpSet) { - SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - memset(pEpSet, 0, sizeof(*pEpSet)); - return; - } - ASSERT(rid == pSyncNode->rid); +void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) { pEpSet->numOfEps = 0; - for (int32_t i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) { - snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn); - pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; - (pEpSet->numOfEps)++; - sInfo("vgId:%d, sync get epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); - } - pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex; - sInfo("vgId:%d, sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); - syncNodeRelease(pSyncNode); -} -#endif - -void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) { SSyncNode* pSyncNode = syncNodeAcquire(rid); - if (pSyncNode == NULL) { - memset(pEpSet, 0, sizeof(*pEpSet)); - return; - } + if (pSyncNode == NULL) return; - pEpSet->numOfEps = 0; for (int32_t i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) { - snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn); - pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; - (pEpSet->numOfEps)++; - sInfo("vgId:%d, sync get retry epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, - pEpSet->eps[i].port); + SEp* pEp = &pEpSet->eps[i]; + tstrncpy(pEp->fqdn, pSyncNode->pRaftCfg->cfg.nodeInfo[i].nodeFqdn, TSDB_FQDN_LEN); + pEp->port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; + pEpSet->numOfEps++; + sInfo("vgId:%d, sync get retry epset, index:%d %s:%d", pSyncNode->vgId, i, pEp->fqdn, pEp->port); } if (pEpSet->numOfEps > 0) { pEpSet->inUse = (pSyncNode->pRaftCfg->cfg.myIndex + 1) % pEpSet->numOfEps; } - sInfo("vgId:%d, sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); + sInfo("vgId:%d, sync get retry epset numOfEps:%d inUse:%d", pSyncNode->vgId, pEpSet->numOfEps, pEpSet->inUse); syncNodeRelease(pSyncNode); } -static void syncGetAndDelRespRpc(SSyncNode* pSyncNode, uint64_t index, SRpcHandleInfo* pInfo) { - SRespStub stub; - int32_t ret = syncRespMgrGetAndDel(pSyncNode->pSyncRespMgr, index, &stub); - if (ret == 1) { - *pInfo = stub.rpcMsg.info; - } - - sTrace("vgId:%d, get seq:%" PRIu64 " rpc handle:%p", pSyncNode->vgId, index, pInfo->handle); -} - int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) { SSyncNode* pSyncNode = syncNodeAcquire(rid); if (pSyncNode == NULL) { @@ -2759,7 +2672,7 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde .flag = flag, }; - syncGetAndDelRespRpc(ths, cbMeta.seqNum, &rpcMsg.info); + syncRespMgrGetAndDel(ths->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info); ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, &cbMeta); } } diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index 56d5e944b85e33c43667a73740ec924f816bc23d..de8f1927ae4d4d1afe0ce7b744f35796d315157a 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -13,21 +13,22 @@ * along with this program. If not, see . */ +#define _DEFAULT_SOURCE #include "syncRespMgr.h" #include "syncRaftEntry.h" #include "syncRaftStore.h" SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) { - SSyncRespMgr *pObj = (SSyncRespMgr *)taosMemoryMalloc(sizeof(SSyncRespMgr)); + SSyncRespMgr *pObj = taosMemoryCalloc(1, sizeof(SSyncRespMgr)); if (pObj == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - memset(pObj, 0, sizeof(SSyncRespMgr)); pObj->pRespHash = taosHashInit(sizeof(uint64_t), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); - ASSERT(pObj->pRespHash != NULL); + if (pObj->pRespHash == NULL) return NULL; + pObj->ttl = ttl; pObj->data = data; pObj->seqNum = 0; @@ -38,93 +39,84 @@ SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) { void syncRespMgrDestroy(SSyncRespMgr *pObj) { if (pObj != NULL) { - taosThreadMutexLock(&(pObj->mutex)); + taosThreadMutexLock(&pObj->mutex); taosHashCleanup(pObj->pRespHash); - taosThreadMutexUnlock(&(pObj->mutex)); + taosThreadMutexUnlock(&pObj->mutex); taosThreadMutexDestroy(&(pObj->mutex)); taosMemoryFree(pObj); } } -int64_t syncRespMgrAdd(SSyncRespMgr *pObj, SRespStub *pStub) { - taosThreadMutexLock(&(pObj->mutex)); +uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub) { + taosThreadMutexLock(&pObj->mutex); - uint64_t keyCode = ++(pObj->seqNum); - taosHashPut(pObj->pRespHash, &keyCode, sizeof(keyCode), pStub, sizeof(SRespStub)); + uint64_t seq = ++(pObj->seqNum); + int32_t code = taosHashPut(pObj->pRespHash, &seq, sizeof(uint64_t), pStub, sizeof(SRespStub)); + sNTrace(pObj->data, "save message handle:%p, type:%s seq:%" PRIu64 " code:0x%x", pStub->rpcMsg.info.handle, + TMSG_INFO(pStub->rpcMsg.msgType), seq, code); - 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; + taosThreadMutexUnlock(&pObj->mutex); + return seq; } -int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t index) { - taosThreadMutexLock(&(pObj->mutex)); +int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq) { + taosThreadMutexLock(&pObj->mutex); - taosHashRemove(pObj->pRespHash, &index, sizeof(index)); + int32_t code = taosHashRemove(pObj->pRespHash, &seq, sizeof(seq)); + sNTrace(pObj->data, "remove message handle, seq:%" PRIu64 " code:%d", seq, code); - taosThreadMutexUnlock(&(pObj->mutex)); - return 0; + taosThreadMutexUnlock(&pObj->mutex); + return code; } -int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub) { - taosThreadMutexLock(&(pObj->mutex)); +int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub) { + taosThreadMutexLock(&pObj->mutex); - void *pTmp = taosHashGet(pObj->pRespHash, &index, sizeof(index)); + SRespStub *pTmp = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t)); if (pTmp != NULL) { memcpy(pStub, pTmp, sizeof(SRespStub)); + sNTrace(pObj->data, "get message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType), seq, + pStub->rpcMsg.info.handle); - 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)); + taosThreadMutexUnlock(&pObj->mutex); return 1; // get one object } - taosThreadMutexUnlock(&(pObj->mutex)); + + taosThreadMutexUnlock(&pObj->mutex); return 0; // get none object } -int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub) { - taosThreadMutexLock(&(pObj->mutex)); +int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo) { + taosThreadMutexLock(&pObj->mutex); - void *pTmp = taosHashGet(pObj->pRespHash, &index, sizeof(index)); - if (pTmp != NULL) { - memcpy(pStub, pTmp, sizeof(SRespStub)); + SRespStub *pStub = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t)); + if (pStub != NULL) { + *pInfo = pStub->rpcMsg.info; + sNTrace(pObj->data, "get-and-del message handle:%p, type:%s seq:%" PRIu64, pStub->rpcMsg.info.handle, + TMSG_INFO(pStub->rpcMsg.msgType), seq); + taosHashRemove(pObj->pRespHash, &seq, sizeof(uint64_t)); - 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)); + taosThreadMutexUnlock(&pObj->mutex); return 1; // get one object } - taosThreadMutexUnlock(&(pObj->mutex)); - return 0; // get none object -} -void syncRespCleanRsp(SSyncRespMgr *pObj) { - taosThreadMutexLock(&(pObj->mutex)); - syncRespCleanByTTL(pObj, -1, true); - taosThreadMutexUnlock(&(pObj->mutex)); -} - -void syncRespClean(SSyncRespMgr *pObj) { - taosThreadMutexLock(&(pObj->mutex)); - syncRespCleanByTTL(pObj, pObj->ttl, false); - taosThreadMutexUnlock(&(pObj->mutex)); + taosThreadMutexUnlock(&pObj->mutex); + return 0; // get none object } -void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { +static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL); int cnt = 0; int sum = 0; SSyncNode *pSyncNode = pObj->data; - SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t)); - ASSERT(delIndexArray != NULL); - sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId); + SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t)); + if (delIndexArray == NULL) return; + sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId); while (pStub) { size_t len; - void * key = taosHashGetKey(pStub, &len); + void *key = taosHashGetKey(pStub, &len); uint64_t *pSeqNum = (uint64_t *)key; sum++; @@ -149,15 +141,15 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { pStub->rpcMsg.contLen = 0; // TODO: and make rpcMsg body, call commit cb - // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta); + // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &pStub->rpcMsg, cbMeta); pStub->rpcMsg.code = TSDB_CODE_SYN_NOT_LEADER; if (pStub->rpcMsg.info.handle != NULL) { - tmsgSendRsp(&(pStub->rpcMsg)); + tmsgSendRsp(&pStub->rpcMsg); } } - pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub); + pStub = taosHashIterate(pObj->pRespHash, pStub); } int32_t arraySize = taosArrayGetSize(delIndexArray); @@ -170,3 +162,15 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { } taosArrayDestroy(delIndexArray); } + +void syncRespCleanRsp(SSyncRespMgr *pObj) { + taosThreadMutexLock(&pObj->mutex); + syncRespCleanByTTL(pObj, -1, true); + taosThreadMutexUnlock(&pObj->mutex); +} + +void syncRespClean(SSyncRespMgr *pObj) { + taosThreadMutexLock(&pObj->mutex); + syncRespCleanByTTL(pObj, pObj->ttl, false); + taosThreadMutexUnlock(&pObj->mutex); +}