From ace84638b8e0f01a0c5c54fb2f5a9bf135a132f0 Mon Sep 17 00:00:00 2001 From: slzhou Date: Thu, 20 Oct 2022 15:30:13 +0800 Subject: [PATCH] fix: add udf call and uv synchronization with uv_mutex --- source/libs/function/src/tudf.c | 65 ++++++++++++++++++--------------- 1 file changed, 36 insertions(+), 29 deletions(-) diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index 69e58829a0..f3a724d4da 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -339,10 +339,11 @@ typedef struct SUdfcProxy { uv_mutex_t udfStubsMutex; SArray *udfStubs; // SUdfcFuncStub + uv_mutex_t udfcUvMutex; int8_t initialized; } SUdfcProxy; -SUdfcProxy gUdfdProxy = {0}; +SUdfcProxy gUdfcProxy = {0}; typedef struct SUdfcUvSession { SUdfcProxy *udfc; @@ -896,23 +897,23 @@ int compareUdfcFuncSub(const void *elem1, const void *elem2) { int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) { int32_t code = 0; - uv_mutex_lock(&gUdfdProxy.udfStubsMutex); + uv_mutex_lock(&gUdfcProxy.udfStubsMutex); SUdfcFuncStub key = {0}; strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN); - int32_t stubIndex = taosArraySearchIdx(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ); + int32_t stubIndex = taosArraySearchIdx(gUdfcProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ); if (stubIndex != -1) { - SUdfcFuncStub *foundStub = taosArrayGet(gUdfdProxy.udfStubs, stubIndex); + SUdfcFuncStub *foundStub = taosArrayGet(gUdfcProxy.udfStubs, stubIndex); UdfcFuncHandle handle = foundStub->handle; if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) { *pHandle = foundStub->handle; ++foundStub->refCount; foundStub->lastRefTime = taosGetTimestampUs(); - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); return 0; } else { fnInfo("invalid handle for %s, refCount: %d, last ref time: %" PRId64 ". remove it from cache", udfName, foundStub->refCount, foundStub->lastRefTime); - taosArrayRemove(gUdfdProxy.udfStubs, stubIndex); + taosArrayRemove(gUdfcProxy.udfStubs, stubIndex); } } *pHandle = NULL; @@ -923,46 +924,46 @@ int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) { stub.handle = *pHandle; ++stub.refCount; stub.lastRefTime = taosGetTimestampUs(); - taosArrayPush(gUdfdProxy.udfStubs, &stub); - taosArraySort(gUdfdProxy.udfStubs, compareUdfcFuncSub); + taosArrayPush(gUdfcProxy.udfStubs, &stub); + taosArraySort(gUdfcProxy.udfStubs, compareUdfcFuncSub); } else { *pHandle = NULL; } - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); return code; } void releaseUdfFuncHandle(char *udfName) { - uv_mutex_lock(&gUdfdProxy.udfStubsMutex); + uv_mutex_lock(&gUdfcProxy.udfStubsMutex); SUdfcFuncStub key = {0}; strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN); - SUdfcFuncStub *foundStub = taosArraySearch(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ); + SUdfcFuncStub *foundStub = taosArraySearch(gUdfcProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ); if (!foundStub) { - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); return; } if (foundStub->refCount > 0) { --foundStub->refCount; } - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); } int32_t cleanUpUdfs() { - int8_t initialized = atomic_load_8(&gUdfdProxy.initialized); + int8_t initialized = atomic_load_8(&gUdfcProxy.initialized); if (!initialized) { return TSDB_CODE_SUCCESS; } - uv_mutex_lock(&gUdfdProxy.udfStubsMutex); - if (gUdfdProxy.udfStubs == NULL || taosArrayGetSize(gUdfdProxy.udfStubs) == 0) { - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + uv_mutex_lock(&gUdfcProxy.udfStubsMutex); + if (gUdfcProxy.udfStubs == NULL || taosArrayGetSize(gUdfcProxy.udfStubs) == 0) { + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); return TSDB_CODE_SUCCESS; } SArray *udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub)); int32_t i = 0; - while (i < taosArrayGetSize(gUdfdProxy.udfStubs)) { - SUdfcFuncStub *stub = taosArrayGet(gUdfdProxy.udfStubs, i); + while (i < taosArrayGetSize(gUdfcProxy.udfStubs)) { + SUdfcFuncStub *stub = taosArrayGet(gUdfcProxy.udfStubs, i); if (stub->refCount == 0) { fnInfo("tear down udf. udf name: %s, handle: %p, ref count: %d", stub->udfName, stub->handle, stub->refCount); doTeardownUdf(stub->handle); @@ -979,9 +980,9 @@ int32_t cleanUpUdfs() { } ++i; } - taosArrayDestroy(gUdfdProxy.udfStubs); - gUdfdProxy.udfStubs = udfStubs; - uv_mutex_unlock(&gUdfdProxy.udfStubsMutex); + taosArrayDestroy(gUdfcProxy.udfStubs); + gUdfcProxy.udfStubs = udfStubs; + uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); return 0; } @@ -1157,9 +1158,11 @@ void onUdfcPipeClose(uv_handle_t *handle) { QUEUE_REMOVE(&task->procTaskQueue); uv_sem_post(&task->taskSem); } + uv_mutex_lock(&gUdfcProxy.udfcUvMutex); if (conn->session != NULL) { conn->session->udfUvPipe = NULL; } + uv_mutex_unlock(&gUdfcProxy.udfcUvMutex); taosMemoryFree(conn->readBuf.buf); taosMemoryFree(conn); taosMemoryFree((uv_pipe_t *)handle); @@ -1553,11 +1556,11 @@ void constructUdfService(void *argsThread) { } int32_t udfcOpen() { - int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 0, 1); + int8_t old = atomic_val_compare_exchange_8(&gUdfcProxy.initialized, 0, 1); if (old == 1) { return 0; } - SUdfcProxy *proxy = &gUdfdProxy; + SUdfcProxy *proxy = &gUdfcProxy; getUdfdPipeName(proxy->udfdPipeName, sizeof(proxy->udfdPipeName)); proxy->udfcState = UDFC_STATE_STARTNG; uv_barrier_init(&proxy->initBarrier, 2); @@ -1567,16 +1570,17 @@ int32_t udfcOpen() { uv_barrier_wait(&proxy->initBarrier); uv_mutex_init(&proxy->udfStubsMutex); proxy->udfStubs = taosArrayInit(8, sizeof(SUdfcFuncStub)); + uv_mutex_init(&proxy->udfcUvMutex); fnInfo("udfc initialized") return 0; } int32_t udfcClose() { - int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 1, 0); + int8_t old = atomic_val_compare_exchange_8(&gUdfcProxy.initialized, 1, 0); if (old == 0) { return 0; } - SUdfcProxy *udfc = &gUdfdProxy; + SUdfcProxy *udfc = &gUdfcProxy; udfc->udfcState = UDFC_STATE_STOPPING; uv_async_send(&udfc->loopStopAsync); uv_thread_join(&udfc->loopThread); @@ -1584,6 +1588,7 @@ int32_t udfcClose() { uv_barrier_destroy(&udfc->initBarrier); taosArrayDestroy(udfc->udfStubs); uv_mutex_destroy(&udfc->udfStubsMutex); + uv_mutex_destroy(&udfc->udfcUvMutex); udfc->udfcState = UDFC_STATE_INITAL; fnInfo("udfc is cleaned up"); return 0; @@ -1611,13 +1616,13 @@ int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) { } int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) { - if (gUdfdProxy.udfcState != UDFC_STATE_READY) { + if (gUdfcProxy.udfcState != UDFC_STATE_READY) { return TSDB_CODE_UDF_INVALID_STATE; } SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask)); task->errCode = 0; task->session = taosMemoryCalloc(1, sizeof(SUdfcUvSession)); - task->session->udfc = &gUdfdProxy; + task->session->udfc = &gUdfcProxy; task->type = UDF_TASK_SETUP; SUdfSetupRequest *req = &task->_setup.req; @@ -1625,7 +1630,7 @@ int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) { int32_t errCode = udfcRunUdfUvTask(task, UV_TASK_CONNECT); if (errCode != 0) { - fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName); + fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfcProxy)->udfdPipeName); taosMemoryFree(task->session); taosMemoryFree(task); return TSDB_CODE_UDF_PIPE_CONNECT_ERR; @@ -1799,10 +1804,12 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) { fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle); // TODO: synchronization refactor between libuv event loop and request thread + uv_mutex_lock(&gUdfcProxy.udfcUvMutex); if (session->udfUvPipe != NULL && session->udfUvPipe->data != NULL) { SClientUvConn *conn = session->udfUvPipe->data; conn->session = NULL; } + uv_mutex_unlock(&gUdfcProxy.udfcUvMutex); taosMemoryFree(session); taosMemoryFree(task);