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

enhance: udf code refactoring

This commit is contained in:
shenglian zhou 2022-05-22 16:08:39 +08:00
parent 317db70052
commit 712217ab02
1 changed files with 4 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -24,7 +24,6 @@
#include "builtinsimpl.h" #include "builtinsimpl.h"
#include "functionMgt.h" #include "functionMgt.h"
//TODO: add unit test
typedef struct SUdfdData { typedef struct SUdfdData {
bool startCalled; bool startCalled;
bool needCleanUp; bool needCleanUp;
@ -45,14 +44,15 @@ typedef struct SUdfdData {
SUdfdData udfdGlobal = {0}; SUdfdData udfdGlobal = {0};
int32_t udfStartUdfd(int32_t startDnodeId);
int32_t udfStopUdfd();
static int32_t udfSpawnUdfd(SUdfdData *pData); static int32_t udfSpawnUdfd(SUdfdData *pData);
void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal); void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal);
static int32_t udfSpawnUdfd(SUdfdData* pData); static int32_t udfSpawnUdfd(SUdfdData* pData);
static void udfUdfdCloseWalkCb(uv_handle_t* handle, void* arg); static void udfUdfdCloseWalkCb(uv_handle_t* handle, void* arg);
static void udfUdfdStopAsyncCb(uv_async_t *async); static void udfUdfdStopAsyncCb(uv_async_t *async);
static void udfWatchUdfd(void *args); static void udfWatchUdfd(void *args);
int32_t udfStartUdfd(int32_t startDnodeId);
int32_t udfStopUdfd();
void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) { void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) {
fnInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal); fnInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
@ -1191,18 +1191,15 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *
switch (task->type) { switch (task->type) {
case UDF_TASK_SETUP: { case UDF_TASK_SETUP: {
//TODO: copy or not
task->_setup.rsp = rsp.setupRsp; task->_setup.rsp = rsp.setupRsp;
break; break;
} }
case UDF_TASK_CALL: { case UDF_TASK_CALL: {
task->_call.rsp = rsp.callRsp; task->_call.rsp = rsp.callRsp;
//TODO: copy or not
break; break;
} }
case UDF_TASK_TEARDOWN: { case UDF_TASK_TEARDOWN: {
task->_teardown.rsp = rsp.teardownRsp; task->_teardown.rsp = rsp.teardownRsp;
//TODO: copy or not?
break; break;
} }
default: { default: {
@ -1398,7 +1395,7 @@ int32_t udfcCreateUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN
request.teardown = task->_teardown.req; request.teardown = task->_teardown.req;
request.type = UDF_TASK_TEARDOWN; request.type = UDF_TASK_TEARDOWN;
} else { } else {
//TODO log and return error fnError("udfc create uv task, invalid task type : %d", task->type);
} }
int32_t bufLen = encodeUdfRequest(NULL, &request); int32_t bufLen = encodeUdfRequest(NULL, &request);
request.msgLen = bufLen; request.msgLen = bufLen;