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

Merge pull request #12001 from taosdata/feature/refator_retry 

refactor(rpc): refactor timeout
This commit is contained in:
Yihao Deng 2022-04-30 10:28:06 +08:00 committed by GitHub
parent 6c178dc0b7 002580dfb2
commit 7aa8cee6a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/index/inc/indexUtil.h
View File

@ -68,7 +68,7 @@ extern "C" {
*/
void iIntersection(SArray *interResults, SArray *finalResult);
/* multi sorted result intersection
/* multi sorted result union
* input: [1, 2, 4, 5]
* [2, 3, 4, 5]
* [1, 4, 5]
@ -76,7 +76,7 @@ void iIntersection(SArray *interResults, SArray *finalResult);
*/
void iUnion(SArray *interResults, SArray *finalResult);
/* sorted array
/* see example
* total: [1, 2, 4, 5, 7, 8]
* except: [4, 5]
* return: [1, 2, 7, 8] saved in total

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

@ -891,7 +891,6 @@ static void doDelayTask(void* param) {
SCliMsg* pMsg = arg->param1;
SCliThrdObj* pThrd = arg->param2;
cliHandleReq(pMsg, pThrd);
taosMemoryFree(arg);

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

@ -425,10 +425,19 @@ void transDQDestroy(SDelayQueue* queue) {
}
int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs) {
uint64_t now = taosGetTimestampMs();
SDelayTask* task = taosMemoryCalloc(1, sizeof(SDelayTask));
task->func = func;
task->arg = arg;
task->execTime = taosGetTimestampMs() + timeoutMs;
task->execTime = now + timeoutMs;
HeapNode* minNode = heapMin(queue->heap);
if (minNode) {
SDelayTask* minTask = container_of(minNode, SDelayTask, node);
if (minTask->execTime < task->execTime) {
timeoutMs = minTask->execTime <= now ? 0 : now - minTask->execTime;
}
}
tTrace("timer %p put task into queue, timeoutMs: %" PRIu64 "", queue->timer, timeoutMs);
heapInsert(queue->heap, &task->node);

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

@ -810,12 +810,14 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj));
thrd->pTransInst = shandle;
thrd->quit = false;
srv->pThreadObj[i] = thrd;
thrd->pTransInst = shandle;
srv->pipe[i] = (uv_pipe_t*)taosMemoryCalloc(2, sizeof(uv_pipe_t));
uv_os_sock_t fds[2];
if (uv_socketpair(SOCK_STREAM, 0, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) {
goto End;