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

Merge pull request #12967 from taosdata/enh/defaultEpSet 

enh: default ep set
This commit is contained in:
Yihao Deng 2022-05-27 13:33:57 +08:00 committed by GitHub
parent abcb0f5f91 6e6dadf5d7
commit a939bc8777
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 82 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/transport/trpc.h
View File

@ -124,6 +124,7 @@ void rpcSendRedirectRsp(void *pConn, const SEpSet *pEpSet);
void rpcSendRequestWithCtx(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid, SRpcCtx *ctx);
int32_t rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
void rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
void rpcSetDefaultAddr(void *thandle, const char *ip, const char *fqdn);
#ifdef __cplusplus
}

14
source/client/src/clientMsgHandler.c
View File

@ -58,7 +58,12 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
return code;
}
if (connectRsp.dnodeNum > 1 && !isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {
if (connectRsp.dnodeNum == 1) {
SEpSet srcEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
SEpSet dstEpSet = connectRsp.epSet;
rpcSetDefaultAddr(pTscObj->pAppInfo->pTransporter, srcEpSet.eps[srcEpSet.inUse].fqdn,
dstEpSet.eps[dstEpSet.inUse].fqdn);
} else if (connectRsp.dnodeNum > 1 && !isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {
updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &connectRsp.epSet);
}
@ -126,9 +131,10 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
if (usedbRsp.vgVersion >= 0) {
uint64_t clusterId = pRequest->pTscObj->pAppInfo->clusterId;
int32_t code1 = catalogGetHandle(clusterId, &pCatalog);
int32_t code1 = catalogGetHandle(clusterId, &pCatalog);
if (code1 != TSDB_CODE_SUCCESS) {
tscWarn("0x%" PRIx64 "catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->requestId, clusterId, tstrerror(code1));
tscWarn("0x%" PRIx64 "catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->requestId, clusterId,
tstrerror(code1));
} else {
catalogRemoveDB(pCatalog, usedbRsp.db, usedbRsp.uid);
}
@ -158,7 +164,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
if (output.dbVgroup) taosHashCleanup(output.dbVgroup->vgHash);
taosMemoryFreeClear(output.dbVgroup);
tscError("0x%" PRIx64" failed to build use db output since %s", pRequest->requestId, terrstr());
tscError("0x%" PRIx64 " failed to build use db output since %s", pRequest->requestId, terrstr());
} else if (output.dbVgroup) {
struct SCatalog* pCatalog = NULL;

11
source/libs/transport/inc/transComm.h
View File

@ -104,6 +104,13 @@ typedef SRpcCtxVal STransCtxVal;
typedef SRpcInfo STrans;
typedef SRpcConnInfo STransHandleInfo;
/*convet from fqdn to ip */
typedef struct SCvtAddr {
char ip[TSDB_FQDN_LEN];
char fqdn[TSDB_FQDN_LEN];
bool cvt;
} SCvtAddr;
typedef struct {
SEpSet epSet; // ip list provided by app
void* ahandle; // handle provided by app
@ -115,6 +122,7 @@ typedef struct {
STransCtx appCtx; //
STransMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API
SCvtAddr cvtAddr;
int hThrdIdx;
} STransConnCtx;
@ -155,7 +163,7 @@ typedef struct {
#pragma pack(pop)
typedef enum { Normal, Quit, Release, Register } STransMsgType;
typedef enum { Normal, Quit, Release, Register, Update } STransMsgType;
typedef enum { ConnNormal, ConnAcquire, ConnRelease, ConnBroken, ConnInPool } ConnStatus;
#define container_of(ptr, type, member) ((type*)((char*)(ptr)-offsetof(type, member)))
@ -231,6 +239,7 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pMsg, STransM
void transSendResponse(const STransMsg* msg);
void transRegisterMsg(const STransMsg* msg);
int transGetConnInfo(void* thandle, STransHandleInfo* pInfo);
void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn);
void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle);
void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle);

5
source/libs/transport/src/trans.c
View File

@ -154,6 +154,11 @@ void rpcReleaseHandle(void* handle, int8_t type) {
(*transReleaseHandle[type])(handle);
}
void rpcSetDefaultAddr(void* thandle, const char* ip, const char* fqdn) {
// later
transSetDefaultAddr(thandle, ip, fqdn);
}
int32_t rpcInit() {
// impl later
return 0;

59
source/libs/transport/src/transCli.c
View File

@ -63,7 +63,10 @@ typedef struct SCliThrdObj {
SDelayQueue* delayQueue;
uint64_t nextTimeout; // next timeout
void* pTransInst; //
bool quit;
SCvtAddr cvtAddr;
bool quit;
} SCliThrdObj;
typedef struct SCliObj {
@ -103,6 +106,7 @@ static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle o
static void cliDestroy(uv_handle_t* handle);
static void cliSend(SCliConn* pConn);
void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr);
/*
* set TCP connection timeout per-socket level
*/
@ -116,7 +120,9 @@ static void cliHandleExcept(SCliConn* conn);
static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrdObj* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease};
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrdObj* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease,
NULL, cliHandleUpdate};
static void cliSendQuit(SCliThrdObj* thrd);
static void destroyUserdata(STransMsg* userdata);
@ -697,6 +703,12 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
transUnrefCliHandle(conn);
}
}
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd) {
STransConnCtx* pCtx = pMsg->ctx;
pThrd->cvtAddr = pCtx->cvtAddr;
destroyCmsg(pMsg);
}
SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) {
SCliConn* conn = NULL;
@ -716,7 +728,17 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) {
}
return conn;
}
void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) {
if (pCvtAddr->cvt == false) {
return;
}
for (int i = 0; i < pEpSet->numOfEps && pEpSet->numOfEps == 1; i++) {
if (strncmp(pEpSet->eps[i].fqdn, pCvtAddr->fqdn, TSDB_FQDN_LEN) == 0) {
memset(pEpSet->eps[i].fqdn, 0, TSDB_FQDN_LEN);
memcpy(pEpSet->eps[i].fqdn, pCvtAddr->ip, TSDB_FQDN_LEN);
}
}
}
void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
uint64_t et = taosGetTimestampUs();
uint64_t el = et - pMsg->st;
@ -726,6 +748,8 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
STransConnCtx* pCtx = pMsg->ctx;
STrans* pTransInst = pThrd->pTransInst;
cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr);
SCliConn* conn = cliGetConn(pMsg, pThrd);
if (conn != NULL) {
conn->hThrdIdx = pCtx->hThrdIdx;
@ -855,7 +879,6 @@ static SCliThrdObj* createThrdObj() {
pThrd->timer.data = pThrd;
pThrd->pool = createConnPool(4);
transDQCreate(pThrd->loop, &pThrd->delayQueue);
pThrd->quit = false;
@ -1088,4 +1111,32 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM
taosMemoryFree(pSem);
}
/*
*
**/
void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) {
STrans* pTransInst = ahandle;
SCvtAddr cvtAddr = {0};
if (ip != NULL && fqdn != NULL) {
memcpy(cvtAddr.ip, ip, strlen(ip));
memcpy(cvtAddr.fqdn, fqdn, strlen(fqdn));
cvtAddr.cvt = true;
}
for (int i = 0; i < pTransInst->numOfThreads; i++) {
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
pCtx->hThrdIdx = i;
pCtx->cvtAddr = cvtAddr;
SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
cliMsg->ctx = pCtx;
cliMsg->type = Update;
SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
tDebug("update epset at thread:%d, threadID:%" PRId64 "", i, thrd->thread);
tsem_t* pSem = pCtx->pSem;
transSendAsync(thrd->asyncPool, &(cliMsg->q));
}
}
#endif

2
source/libs/transport/src/transSrv.c
View File

@ -146,7 +146,7 @@ static void uvHandleRelease(SSrvMsg* msg, SWorkThrdObj* thrd);
static void uvHandleResp(SSrvMsg* msg, SWorkThrdObj* thrd);
static void uvHandleRegister(SSrvMsg* msg, SWorkThrdObj* thrd);
static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
uvHandleRegister};
uvHandleRegister, NULL};
static int32_t exHandlesMgt;