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

add syncGetEpSet

This commit is contained in:
Minghao Li 2022-04-30 10:29:29 +08:00
parent 9432dc6f6c
commit 2584b034df
3 changed files with 30 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
source/dnode/mgmt/implement/src/dmTransport.c
View File

@ -320,6 +320,27 @@ static inline void dmSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
} }
} }
static inline void dmSendRedirectRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp, const SEpSet *pNewEpSet) {
ASSERT(pRsp->code == TSDB_CODE_RPC_REDIRECT);
ASSERT(pRsp->pCont == NULL);
if (pWrapper->procType != DND_PROC_CHILD) {
SRpcMsg resp = {0};
SMEpSet msg = {.epSet = *pNewEpSet};
int32_t len = tSerializeSMEpSet(NULL, 0, &msg);
resp.pCont = rpcMallocCont(len);
resp.contLen = len;
tSerializeSMEpSet(resp.pCont, len, &msg);
resp.code = TSDB_CODE_RPC_REDIRECT;
resp.handle = pRsp->handle;
resp.refId = pRsp->refId;
rpcSendResponse(&resp);
} else {
taosProcPutToParentQ(pWrapper->procObj, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_FUNC_RSP);
}
}
#if 0
static inline void dmSendRedirectRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp, const SEpSet *pNewEpSet) { static inline void dmSendRedirectRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp, const SEpSet *pNewEpSet) {
ASSERT(pRsp->code == TSDB_CODE_RPC_REDIRECT); ASSERT(pRsp->code == TSDB_CODE_RPC_REDIRECT);
if (pWrapper->procType != DND_PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {
@ -328,6 +349,7 @@ static inline void dmSendRedirectRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp
taosProcPutToParentQ(pWrapper->procObj, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_FUNC_RSP); taosProcPutToParentQ(pWrapper->procObj, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_FUNC_RSP);
} }
} }
#endif
static inline void dmRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) { static inline void dmRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
if (pWrapper->procType != DND_PROC_CHILD) { if (pWrapper->procType != DND_PROC_CHILD) {

1
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -155,6 +155,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
rsp.code = TSDB_CODE_RPC_REDIRECT; rsp.code = TSDB_CODE_RPC_REDIRECT;
SEpSet newEpSet; SEpSet newEpSet;
syncGetEpSet(vnodeGetSyncHandle(pVnode->pImpl), &newEpSet); syncGetEpSet(vnodeGetSyncHandle(pVnode->pImpl), &newEpSet);
newEpSet.inUse = (newEpSet.inUse + 1) % newEpSet.numOfEps;
tmsgSendRedirectRsp(&rsp, &newEpSet); tmsgSendRedirectRsp(&rsp, &newEpSet);
} else if (ret == TAOS_SYNC_PROPOSE_OTHER_ERROR) { } else if (ret == TAOS_SYNC_PROPOSE_OTHER_ERROR) {

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

@ -174,11 +174,16 @@ void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
assert(rid == pSyncNode->rid); assert(rid == pSyncNode->rid);
pEpSet->numOfEps = 0; pEpSet->numOfEps = 0;
for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) { for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
snprintf(pEpSet->eps->fqdn, sizeof(pEpSet->eps->fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn); snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn);
pEpSet->eps->port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort;
(pEpSet->numOfEps)++; (pEpSet->numOfEps)++;
sInfo("syncGetEpSet index:%d %s:%d", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
} }
pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex; pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex;
sInfo("syncGetEpSet pEpSet->inUse:%d ", pEpSet->inUse);
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
} }