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

support fail fast

This commit is contained in:
yihaoDeng 2022-12-06 10:30:14 +08:00
parent 4be196491e
commit 9ce9d21507
4 changed files with 16 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -48,6 +48,11 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
return (*msgFp)(pWrapper->pMgmt, pMsg); return (*msgFp)(pWrapper->pMgmt, pMsg);
} }
static bool dmFailFastFp(tmsg_t msgType) {
// add more msg type later
return msgType == TDMT_SYNC_HEARTBEAT;
}
static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
SDnodeTrans *pTrans = &pDnode->trans; SDnodeTrans *pTrans = &pDnode->trans;
int32_t code = -1; int32_t code = -1;
@ -260,6 +265,10 @@ int32_t dmInitClient(SDnode *pDnode) {
rpcInit.retryMaxInterval = tsRedirectMaxPeriod; rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
rpcInit.retryMaxTimouet = tsMaxRetryWaitTime; rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
rpcInit.failFastInterval = 1000; // interval threshold(ms)
rpcInit.failFastThreshold = 3; // failed threshold
rpcInit.ffp = dmFailFastFp;
pTrans->clientRpc = rpcOpen(&rpcInit); pTrans->clientRpc = rpcOpen(&rpcInit);
if (pTrans->clientRpc == NULL) { if (pTrans->clientRpc == NULL) {
dError("failed to init dnode rpc client"); dError("failed to init dnode rpc client");

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

@ -886,7 +886,8 @@ void cliConnCb(uv_connect_t* req, int status) {
tError("%s conn %p failed to connect to %s:%d, reason:%s", CONN_GET_INST_LABEL(pConn), pConn, pConn->ip, tError("%s conn %p failed to connect to %s:%d, reason:%s", CONN_GET_INST_LABEL(pConn), pConn, pConn->ip,
pConn->port, uv_strerror(status)); pConn->port, uv_strerror(status));
SCliMsg* pMsg = transQueueGet(&pConn->cliMsgs, 0); SCliMsg* pMsg = transQueueGet(&pConn->cliMsgs, 0);
if (REQUEST_NO_RESP(&pMsg->msg)) { STrans* pTransInst = pThrd->pTransInst;
if (REQUEST_NO_RESP(&pMsg->msg) && (pTransInst->failFastFp != NULL && pTransInst->failFastFp(pMsg->msg.msgType))) {
char* ip = pConn->ip; char* ip = pConn->ip;
uint32_t port = pConn->port; uint32_t port = pConn->port;
char key[TSDB_FQDN_LEN + 64] = {0}; char key[TSDB_FQDN_LEN + 64] = {0};
@ -895,7 +896,7 @@ void cliConnCb(uv_connect_t* req, int status) {
SFailFastItem* item = taosHashGet(pThrd->failFastCache, key, strlen(key)); SFailFastItem* item = taosHashGet(pThrd->failFastCache, key, strlen(key));
int64_t cTimestamp = taosGetTimestampMs(); int64_t cTimestamp = taosGetTimestampMs();
if (item != NULL) { if (item != NULL) {
if (cTimestamp - item->timestamp < ((STrans*)pThrd->pTransInst)->failFastInterval) { if (cTimestamp - item->timestamp < pTransInst->failFastInterval) {
item->count++; item->count++;
} else { } else {
item->count = 1; item->count = 1;
@ -1057,7 +1058,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
return; return;
} }
if (REQUEST_NO_RESP(&pMsg->msg)) { if (REQUEST_NO_RESP(&pMsg->msg) && (pTransInst->failFastFp != NULL && pTransInst->failFastFp(pMsg->msg.msgType))) {
char* ip = EPSET_GET_INUSE_IP(&pCtx->epSet); char* ip = EPSET_GET_INUSE_IP(&pCtx->epSet);
uint32_t port = EPSET_GET_INUSE_PORT(&pCtx->epSet); uint32_t port = EPSET_GET_INUSE_PORT(&pCtx->epSet);
char key[TSDB_FQDN_LEN + 64] = {0}; char key[TSDB_FQDN_LEN + 64] = {0};
@ -1376,6 +1377,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
taosMemoryFree(pThrd->prepare); taosMemoryFree(pThrd->prepare);
taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd->loop);
taosHashCleanup(pThrd->fqdn2ipCache); taosHashCleanup(pThrd->fqdn2ipCache);
taosHashCleanup(pThrd->failFastCache);
taosMemoryFree(pThrd); taosMemoryFree(pThrd);
} }

2
source/os/src/osTime.c
View File

@ -572,7 +572,7 @@ int32_t taosClockGetTime(int clock_id, struct timespec *pTS) {
offsetInitFinished = true; offsetInitFinished = true;
} else { } else {
while (!offsetInitFinished) while (!offsetInitFinished)
; // Ensure initialization is completed. ; // Ensure initialization is completed.
} }
GetSystemTimeAsFileTime(&f); GetSystemTimeAsFileTime(&f);

2
source/util/src/tlog.c
View File

@ -496,7 +496,7 @@ void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, cons
if (!osLogSpaceAvailable()) return; if (!osLogSpaceAvailable()) return;
if (!(dflag & DEBUG_FILE) && !(dflag & DEBUG_SCREEN)) return; if (!(dflag & DEBUG_FILE) && !(dflag & DEBUG_SCREEN)) return;
char *buffer = taosMemoryMalloc(LOG_MAX_LINE_DUMP_BUFFER_SIZE); char *buffer = taosMemoryMalloc(LOG_MAX_LINE_DUMP_BUFFER_SIZE);
int32_t len = taosBuildLogHead(buffer, flags); int32_t len = taosBuildLogHead(buffer, flags);
va_list argpointer; va_list argpointer;