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

opt transport

This commit is contained in:
yihaoDeng 2024-09-19 09:04:56 +08:00
parent 88c0724468
commit 6d3b8c0594
2 changed files with 28 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -141,28 +141,24 @@ typedef struct {
SEpSet origEpSet;
void* ahandle; // handle provided by app
tmsg_t msgType; // message type
int8_t connType; // connection type cli/srv
STransCtx userCtx; //
STransMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API
STransSyncMsg* pSyncMsg; // for syncchronous with timeout API
int64_t syncMsgRef;
SCvtAddr cvtAddr;
SCvtAddr* pCvtAddr;
int64_t retryInitTimestamp;
int64_t retryNextInterval;
int64_t retryMaxTimeout;
int32_t retryMinInterval;
int32_t retryMaxInterval;
int32_t retryStepFactor;
int64_t retryMaxTimeout;
int64_t retryInitTimestamp;
int64_t retryNextInterval;
bool retryInit;
int32_t retryStep;
int8_t epsetRetryCnt;
int32_t retryCode;
void* task;
int hThrdIdx;
int8_t retryInit;
int8_t epsetRetryCnt;
} SReqCtx;
#pragma pack(push, 1)

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

@ -120,17 +120,13 @@ typedef struct {
typedef struct SCliReq {
SReqCtx* ctx;
STransMsg msg;
queue q;
STransMsgType type;
// int64_t refId;
uint64_t st;
int sent; //(0: no send, 1: alread sent)
queue seqq;
int64_t seq;
int32_t sent; //(0: no send, 1: alread sent)
STransMsg msg;
queue qlist;
} SCliReq;
typedef struct SCliThrd {
@ -152,7 +148,7 @@ typedef struct SCliThrd {
void (*destroyAhandleFp)(void* ahandle);
SHashObj* fqdn2ipCache;
SCvtAddr cvtAddr;
SCvtAddr* pCvtAddr;
SHashObj* failFastCache;
SHashObj* batchCache;
@ -1474,12 +1470,18 @@ static void cliHandleQuit(SCliThrd* pThrd, SCliReq* pReq) {
static void cliHandleRelease(SCliThrd* pThrd, SCliReq* pReq) { return; }
static void cliHandleUpdate(SCliThrd* pThrd, SCliReq* pReq) {
SReqCtx* pCtx = pReq->ctx;
pThrd->cvtAddr = pCtx->cvtAddr;
if (pThrd->pCvtAddr != NULL) {
taosMemoryFreeClear(pThrd->pCvtAddr);
}
pThrd->pCvtAddr = pCtx->pCvtAddr;
destroyReq(pReq);
return;
}
FORCE_INLINE int32_t cliMayCvtFqdnToIp(SEpSet* pEpSet, const SCvtAddr* pCvtAddr) {
if (pCvtAddr == NULL) {
return 0;
}
if (pCvtAddr->cvt == false) {
if (EPSET_IS_VALID(pEpSet)) {
return 0;
@ -2001,7 +2003,7 @@ int32_t initCb(void* thrd, SCliReq* pReq, STransMsg* pResp) {
if (pReq->ctx == NULL) {
return 0;
}
return cliMayCvtFqdnToIp(&pReq->ctx->epSet, &pThrd->cvtAddr);
return cliMayCvtFqdnToIp(&pReq->ctx->epSet, pThrd->pCvtAddr);
}
int32_t notifyExceptCb(void* thrd, SCliReq* pReq, STransMsg* pResp) {
SCliThrd* pThrd = thrd;
@ -2201,6 +2203,8 @@ static void destroyThrdObj(SCliThrd* pThrd) {
taosHashCleanup(pThrd->pIdConnTable);
taosMemoryFree(pThrd->pCvtAddr);
taosMemoryFree(pThrd);
}
@ -2426,7 +2430,7 @@ void cliRetryMayInitCtx(STrans* pInst, SCliReq* pReq) {
pCtx->retryInitTimestamp = taosGetTimestampMs();
pCtx->retryNextInterval = pCtx->retryMinInterval;
pCtx->retryStep = 0;
pCtx->retryInit = true;
pCtx->retryInit = 1;
pCtx->retryCode = TSDB_CODE_SUCCESS;
pReq->msg.info.handle = 0;
}
@ -2756,7 +2760,6 @@ static int32_t transInitMsg(void* pInstRef, const SEpSet* pEpSet, STransMsg* pRe
pCliReq->msg = *pReq;
pCliReq->st = taosGetTimestampUs();
pCliReq->type = Normal;
QUEUE_INIT(&pCliReq->seqq);
*pCliMsg = pCliReq;
@ -3023,7 +3026,6 @@ int32_t transSendRecvWithTimeout(void* pInstRef, SEpSet* pEpSet, STransMsg* pReq
pCliReq->msg = *pReq;
pCliReq->st = taosGetTimestampUs();
pCliReq->type = Normal;
// pCliReq->refId = (int64_t)pInstRef;
STraceId* trace = &pReq->info.traceId;
tGDebug("%s send request at thread:%08" PRId64 ", dst:%s:%d, app:%p", transLabel(pInst), pThrd->pid,
@ -3083,10 +3085,12 @@ int32_t transSetDefaultAddr(void* pInstRef, const char* ip, const char* fqdn) {
break;
}
pCtx->cvtAddr = cvtAddr;
pCtx->pCvtAddr = (SCvtAddr*)taosMemoryCalloc(1, sizeof(SCvtAddr));
memcpy(pCtx->pCvtAddr, &cvtAddr, sizeof(SCvtAddr));
SCliReq* pReq = taosMemoryCalloc(1, sizeof(SCliReq));
if (pReq == NULL) {
taosMemoryFree(pCtx->pCvtAddr);
taosMemoryFree(pCtx);
code = terrno;
break;
@ -3099,6 +3103,7 @@ int32_t transSetDefaultAddr(void* pInstRef, const char* ip, const char* fqdn) {
tDebug("%s update epset at thread:%08" PRId64, pInst->label, thrd->pid);
if ((code = transAsyncSend(thrd->asyncPool, &(pReq->q))) != 0) {
taosMemoryFree(pCtx->pCvtAddr);
destroyReq(pReq);
if (code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT) {
code = TSDB_CODE_RPC_MODULE_QUIT;