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

Merge pull request #12488 from taosdata/feature/udf 

fix: clean the udf handles cache when pipe error
This commit is contained in:
shenglian-zhou 2022-05-14 23:53:54 +08:00 committed by GitHub
parent a170b3fa9d 692ae29c6c
commit 1e385277b1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -825,6 +825,11 @@ void onUdfcPipeClose(uv_handle_t *handle) {
taosMemoryFree(conn->readBuf.buf); taosMemoryFree(conn->readBuf.buf);
taosMemoryFree(conn); taosMemoryFree(conn);
taosMemoryFree((uv_pipe_t *) handle); taosMemoryFree((uv_pipe_t *) handle);
//clear the udf handles cache
uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
taosArrayClear(gUdfdProxy.udfStubs);
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
} }
int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask) { int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask) {
@ -1140,7 +1145,7 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
return code; return code;
} }
void udfClientAsyncCb(uv_async_t *async) { void udfcAsyncTaskCb(uv_async_t *async) {
SUdfcProxy *udfc = async->data; SUdfcProxy *udfc = async->data;
QUEUE wq; QUEUE wq;
@ -1204,7 +1209,7 @@ void constructUdfService(void *argsThread) {
SUdfcProxy *udfc = (SUdfcProxy *)argsThread; SUdfcProxy *udfc = (SUdfcProxy *)argsThread;
uv_loop_init(&udfc->uvLoop); uv_loop_init(&udfc->uvLoop);
uv_async_init(&udfc->uvLoop, &udfc->loopTaskAync, udfClientAsyncCb); uv_async_init(&udfc->uvLoop, &udfc->loopTaskAync, udfcAsyncTaskCb);
udfc->loopTaskAync.data = udfc; udfc->loopTaskAync.data = udfc;
uv_async_init(&udfc->uvLoop, &udfc->loopStopAsync, udfStopAsyncCb); uv_async_init(&udfc->uvLoop, &udfc->loopStopAsync, udfStopAsyncCb);
udfc->loopStopAsync.data = udfc; udfc->loopStopAsync.data = udfc;
@ -1472,6 +1477,7 @@ int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols,
return code; return code;
} }
//TODO: when to teardown udf. teardown udf is not called
int32_t doTeardownUdf(UdfcFuncHandle handle) { int32_t doTeardownUdf(UdfcFuncHandle handle) {
fnInfo("tear down udf. udf func handle: %p", handle); fnInfo("tear down udf. udf func handle: %p", handle);