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

enhance: refactor cleanup udf function

This commit is contained in:
shenglian zhou 2023-04-17 10:34:34 +08:00
parent 430457e5f8
commit 1a0c9f31bc
1 changed files with 42 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -968,6 +968,8 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo *pRes
int32_t udfAggProcess(struct SqlFunctionCtx *pCtx); int32_t udfAggProcess(struct SqlFunctionCtx *pCtx);
int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock); int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock);
void cleanupNotExpiredUdfs();
void cleanupExpiredUdfs();
int compareUdfcFuncSub(const void *elem1, const void *elem2) { int compareUdfcFuncSub(const void *elem1, const void *elem2) {
SUdfcFuncStub *stub1 = (SUdfcFuncStub *)elem1; SUdfcFuncStub *stub1 = (SUdfcFuncStub *)elem1;
SUdfcFuncStub *stub2 = (SUdfcFuncStub *)elem2; SUdfcFuncStub *stub2 = (SUdfcFuncStub *)elem2;
@ -1040,18 +1042,32 @@ void releaseUdfFuncHandle(char *udfName, UdfcFuncHandle handle) {
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
} }
int32_t cleanUpUdfs() { void cleanupExpiredUdfs() {
int8_t initialized = atomic_load_8(&gUdfcProxy.initialized); int32_t i = 0;
if (!initialized) { SArray *expiredUdfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub));
return TSDB_CODE_SUCCESS; while (i < taosArrayGetSize(gUdfcProxy.expiredUdfStubs)) {
SUdfcFuncStub *stub = taosArrayGet(gUdfcProxy.expiredUdfStubs, i);
if (stub->refCount == 0) {
fnInfo("tear down udf. expired. udf name: %s, handle: %p, ref count: %d", stub->udfName, stub->handle, stub->refCount);
doTeardownUdf(stub->handle);
} else {
fnInfo("udf still in use. expired. udf name: %s, ref count: %d, create time: %" PRId64 ", handle: %p", stub->udfName,
stub->refCount, stub->createTime, stub->handle);
UdfcFuncHandle handle = stub->handle;
if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) {
taosArrayPush(expiredUdfStubs, stub);
} else {
fnInfo("udf invalid handle for %s, expired. refCount: %d, create time: %" PRId64 ". remove it from cache",
stub->udfName, stub->refCount, stub->createTime);
} }
}
++i;
}
taosArrayDestroy(gUdfcProxy.expiredUdfStubs);
gUdfcProxy.expiredUdfStubs = expiredUdfStubs;
}
uv_mutex_lock(&gUdfcProxy.udfStubsMutex); void cleanupNotExpiredUdfs() {
if ((gUdfcProxy.udfStubs == NULL || taosArrayGetSize(gUdfcProxy.udfStubs) == 0) &&
(gUdfcProxy.expiredUdfStubs == NULL || taosArrayGetSize(gUdfcProxy.expiredUdfStubs) == 0)) {
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return TSDB_CODE_SUCCESS;
}
SArray *udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub)); SArray *udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub));
int32_t i = 0; int32_t i = 0;
while (i < taosArrayGetSize(gUdfcProxy.udfStubs)) { while (i < taosArrayGetSize(gUdfcProxy.udfStubs)) {
@ -1074,28 +1090,24 @@ int32_t cleanUpUdfs() {
} }
taosArrayDestroy(gUdfcProxy.udfStubs); taosArrayDestroy(gUdfcProxy.udfStubs);
gUdfcProxy.udfStubs = udfStubs; gUdfcProxy.udfStubs = udfStubs;
}
SArray *expiredUdfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub)); int32_t cleanUpUdfs() {
while (i < taosArrayGetSize(gUdfcProxy.expiredUdfStubs)) { int8_t initialized = atomic_load_8(&gUdfcProxy.initialized);
SUdfcFuncStub *stub = taosArrayGet(gUdfcProxy.expiredUdfStubs, i); if (!initialized) {
if (stub->refCount == 0) { return TSDB_CODE_SUCCESS;
fnInfo("tear down udf. expired. udf name: %s, handle: %p, ref count: %d", stub->udfName, stub->handle, stub->refCount);
doTeardownUdf(stub->handle);
} else {
fnInfo("udf still in use. expired. udf name: %s, ref count: %d, create time: %" PRId64 ", handle: %p", stub->udfName,
stub->refCount, stub->createTime, stub->handle);
UdfcFuncHandle handle = stub->handle;
if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) {
taosArrayPush(expiredUdfStubs, stub);
} else {
fnInfo("udf invalid handle for %s, expired. refCount: %d, create time: %" PRId64 ". remove it from cache",
stub->udfName, stub->refCount, stub->createTime);
} }
uv_mutex_lock(&gUdfcProxy.udfStubsMutex);
if ((gUdfcProxy.udfStubs == NULL || taosArrayGetSize(gUdfcProxy.udfStubs) == 0) &&
(gUdfcProxy.expiredUdfStubs == NULL || taosArrayGetSize(gUdfcProxy.expiredUdfStubs) == 0)) {
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return TSDB_CODE_SUCCESS;
} }
++i;
} cleanupNotExpiredUdfs();
taosArrayDestroy(gUdfcProxy.expiredUdfStubs); cleanupExpiredUdfs();
gUdfcProxy.expiredUdfStubs = expiredUdfStubs;
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex); uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return 0; return 0;
} }