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

refactor(rpc): fefactor retry way

This commit is contained in:
yihaoDeng 2022-04-28 11:56:00 +08:00
parent 4e7e83399f
commit 9af1206cbd
3 changed files with 70 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -329,6 +329,14 @@ void transQueueClear(STransQueue* queue);
*/
void transQueueDestroy(STransQueue* queue);
/*
* delay queue based on uv loop and uv timer, and only used in retry
*/
typedef struct STaskArg {
void* param1;
void* param2;
} STaskArg;
typedef struct SDelayTask {
void (*func)(void* arg);
void* arg;

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

@ -60,7 +60,7 @@ typedef struct SCliThrdObj {
// msg queue
queue msg;
TdThreadMutex msgMtx;
SDelayQueue* delayQueue;
uint64_t nextTimeout; // next timeout
void* pTransInst; //
bool quit;
@ -838,12 +838,13 @@ static SCliThrdObj* createThrdObj() {
uv_loop_init(pThrd->loop);
pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 5, pThrd, cliAsyncCb);
uv_timer_init(pThrd->loop, &pThrd->timer);
pThrd->timer.data = pThrd;
pThrd->pool = createConnPool(4);
transCreateDelayQueue(pThrd->loop, &pThrd->delayQueue);
pThrd->quit = false;
return pThrd;
}
@ -851,12 +852,13 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
if (pThrd == NULL) {
return;
}
taosThreadJoin(pThrd->thread, NULL);
CLI_RELEASE_UV(pThrd->loop);
taosThreadMutexDestroy(&pThrd->msgMtx);
transDestroyAsyncPool(pThrd->asyncPool);
uv_timer_stop(&pThrd->timer);
transDestroyDelayQueue(pThrd->delayQueue);
taosMemoryFree(pThrd->loop);
taosMemoryFree(pThrd);
}
@ -885,6 +887,16 @@ int cliRBChoseIdx(STrans* pTransInst) {
}
return index % pTransInst->numOfThreads;
}
static void doDelayTask(void* param) {
STaskArg* arg = param;
SCliMsg* pMsg = arg->param1;
SCliThrdObj* pThrd = arg->param2;
cliHandleReq(pMsg, pThrd);
taosMemoryFree(arg);
}
int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
SCliThrdObj* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst;
@ -908,7 +920,12 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
if (msgType == TDMT_MND_CONNECT && pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
if (pCtx->retryCount < pEpSet->numOfEps) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
cliHandleReq(pMsg, pThrd);
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transPutTaskToDelayQueue(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
cliDestroy((uv_handle_t*)pConn->stream);
return -1;
}
@ -920,8 +937,11 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
tDeserializeSMEpSet(pResp->pCont, pResp->contLen, &emsg);
pCtx->epSet = emsg.epSet;
}
cliHandleReq(pMsg, pThrd);
// release pConn
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transPutTaskToDelayQueue(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
addConnToPool(pThrd, pConn);
return -1;
}

48
source/libs/transport/src/transComm.c
View File

@ -363,7 +363,7 @@ static int32_t timeCompare(const HeapNode* a, const HeapNode* b) {
SDelayTask* arg1 = container_of(a, SDelayTask, node);
SDelayTask* arg2 = container_of(b, SDelayTask, node);
if (arg1->execTime > arg2->execTime) {
return -1;
return 0;
} else {
return 1;
}
@ -371,24 +371,26 @@ static int32_t timeCompare(const HeapNode* a, const HeapNode* b) {
static void transDelayQueueTimeout(uv_timer_t* timer) {
SDelayQueue* queue = timer->data;
HeapNode* node = heapMin(queue->heap);
if (node == NULL) {
// DO NOTHING
}
heapRemove(queue->heap, node);
SDelayTask* task = container_of(node, SDelayTask, node);
tTrace("timer %p timeout", timer);
uint64_t timeout = 0;
do {
HeapNode* minNode = heapMin(queue->heap);
if (minNode == NULL) break;
SDelayTask* task = container_of(minNode, SDelayTask, node);
if (task->execTime <= taosGetTimestampMs()) {
heapRemove(queue->heap, minNode);
task->func(task->arg);
taosMemoryFree(task);
node = heapMin(queue->heap);
if (node == NULL) {
return;
timeout = 0;
} else {
timeout = task->execTime - taosGetTimestampMs();
break;
}
task = container_of(node, SDelayTask, node);
uint64_t timeout = task->execTime > uv_now(queue->loop) ? task->execTime - uv_now(queue->loop) : 0;
} while (1);
if (timeout != 0) {
uv_timer_start(queue->timer, transDelayQueueTimeout, timeout, 0);
}
}
int transCreateDelayQueue(uv_loop_t* loop, SDelayQueue** queue) {
uv_timer_t* timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
uv_timer_init(loop, timer);
@ -406,8 +408,18 @@ int transCreateDelayQueue(uv_loop_t* loop, SDelayQueue** queue) {
}
void transDestroyDelayQueue(SDelayQueue* queue) {
uv_timer_stop(queue->timer);
taosMemoryFree(queue->timer);
while (heapSize(queue->heap) > 0) {
HeapNode* minNode = heapMin(queue->heap);
if (minNode == NULL) {
return;
}
heapRemove(queue->heap, minNode);
SDelayTask* task = container_of(minNode, SDelayTask, node);
taosMemoryFree(task);
}
heapDestroy(queue->heap);
taosMemoryFree(queue);
}
@ -417,11 +429,11 @@ int transPutTaskToDelayQueue(SDelayQueue* queue, void (*func)(void* arg), void*
task->func = func;
task->arg = arg;
task->execTime = uv_now(queue->loop) + timeoutMs;
task->execTime = taosGetTimestampMs() + timeoutMs;
int size = heapSize(queue->heap);
tTrace("timer %p put task into queue, timeoutMs: %" PRIu64 "", queue->timer, timeoutMs);
heapInsert(queue->heap, &task->node);
if (size == 1) {
if (heapSize(queue->heap) == 1) {
uv_timer_start(queue->timer, transDelayQueueTimeout, timeoutMs, 0);
}