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

fix: finish uv task when event loop start uv task failed

This commit is contained in:
slzhou 2022-05-14 17:20:22 +08:00
parent 4a316eae3f
commit 7161919086
1 changed files with 9 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/libs/function/src/tudf.c
View File

@ -1072,6 +1072,8 @@ int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask) {
int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) { int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
fnTrace("event loop start uv task. task: %d, %p", uvTask->type, uvTask); fnTrace("event loop start uv task. task: %d, %p", uvTask->type, uvTask);
int32_t code = 0;
switch (uvTask->type) { switch (uvTask->type) {
case UV_TASK_CONNECT: { case UV_TASK_CONNECT: {
uv_pipe_t *pipe = taosMemoryMalloc(sizeof(uv_pipe_t)); uv_pipe_t *pipe = taosMemoryMalloc(sizeof(uv_pipe_t));
@ -1091,6 +1093,7 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
uv_connect_t *connReq = taosMemoryMalloc(sizeof(uv_connect_t)); uv_connect_t *connReq = taosMemoryMalloc(sizeof(uv_connect_t));
connReq->data = uvTask; connReq->data = uvTask;
uv_pipe_connect(connReq, pipe, uvTask->udfc->udfdPipeName, onUdfcPipeConnect); uv_pipe_connect(connReq, pipe, uvTask->udfc->udfdPipeName, onUdfcPipeConnect);
code = 0;
break; break;
} }
case UV_TASK_REQ_RSP: { case UV_TASK_REQ_RSP: {
@ -1101,12 +1104,14 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
if (err != 0) { if (err != 0) {
fnError("udfc event loop start req/rsp task uv_write failed. code: %s", uv_strerror(err)); fnError("udfc event loop start req/rsp task uv_write failed. code: %s", uv_strerror(err));
} }
code = err;
break; break;
} }
case UV_TASK_DISCONNECT: { case UV_TASK_DISCONNECT: {
SClientUvConn *conn = uvTask->pipe->data; SClientUvConn *conn = uvTask->pipe->data;
QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue); QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
uv_close((uv_handle_t *) uvTask->pipe, onUdfcPipeClose); uv_close((uv_handle_t *) uvTask->pipe, onUdfcPipeClose);
code = 0;
break; break;
} }
default: { default: {
@ -1115,7 +1120,7 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
} }
} }
return 0; return code;
} }
void udfClientAsyncCb(uv_async_t *async) { void udfClientAsyncCb(uv_async_t *async) {
@ -1133,6 +1138,9 @@ void udfClientAsyncCb(uv_async_t *async) {
int32_t code = udfcStartUvTask(task); int32_t code = udfcStartUvTask(task);
if (code == 0) { if (code == 0) {
QUEUE_INSERT_TAIL(&udfc->uvProcTaskQueue, &task->procTaskQueue); QUEUE_INSERT_TAIL(&udfc->uvProcTaskQueue, &task->procTaskQueue);
} else {
task->errCode = code;
uv_sem_post(&task->taskSem);
} }
} }