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

refactor: rename udfd to taosudf

This commit is contained in:
Simon Guan 2025-03-06 22:20:52 +08:00
parent 1de5e5e79c
commit 6250adb3b1
24 changed files with 178 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/en/07-develop/09-udf.md
View File

@ -495,10 +495,10 @@ taos> select myfun(v1, v2) from t;
DB error: udf function execution failure (0.011088s)
```
Unfortunately, the execution failed. What could be the reason? Check the udfd process logs.
Unfortunately, the execution failed. What could be the reason? Check the taosudf process logs.
```shell
tail -10 /var/log/taos/udfd.log
tail -10 /var/log/taos/taosudf.log
```
Found the following error messages.

8
docs/en/14-reference/09-error-code.md
View File

@ -485,10 +485,10 @@ This document details the server error codes that may be encountered when using
| Error Code | Description | Possible Scenarios or Reasons | Recommended Actions |
| ---------- | ---------------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| 0x80002901 | udf is stopping | udf call received when dnode exits | Stop executing udf queries |
| 0x80002902 | udf pipe read error | Error occurred when taosd reads from udfd pipe | udfd unexpectedly exits, 1) C udf crash 2) udfd crash |
| 0x80002903 | udf pipe connect error | Error establishing pipe connection to udfd in taosd | 1) Corresponding udfd not started in taosd. Restart taosd |
| 0x80002904 | udf pipe not exist | Connection error occurs between two phases of udf setup, call, and teardown, causing the connection to disappear, subsequent phases continue | udfd unexpectedly exits, 1) C udf crash 2) udfd crash |
| 0x80002905 | udf load failure | Error loading udf in udfd | 1) udf does not exist in mnode 2) Error in udf loading. Check logs |
| 0x80002902 | udf pipe read error | Error occurred when taosd reads from taosudf pipe | taosudf unexpectedly exits, 1) C udf crash 2) taosudf crash |
| 0x80002903 | udf pipe connect error | Error establishing pipe connection to taosudf in taosd | 1) Corresponding taosudf not started in taosd. Restart taosd |
| 0x80002904 | udf pipe not exist | Connection error occurs between two phases of udf setup, call, and teardown, causing the connection to disappear, subsequent phases continue | taosudf unexpectedly exits, 1) C udf crash 2) taosudf crash |
| 0x80002905 | udf load failure | Error loading udf in taosudf | 1) udf does not exist in mnode 2) Error in udf loading. Check logs |
| 0x80002906 | udf invalid function input | udf input check | udf function does not accept input, such as wrong column type |
| 0x80002907 | udf invalid bufsize | Intermediate result in udf aggregation function exceeds specified bufsize | Increase bufsize, or reduce intermediate result size |
| 0x80002908 | udf invalid output type | udf output type differs from the type specified when creating udf | Modify udf, or the type when creating udf, to match the result |

4
docs/zh/07-develop/09-udf.md
View File

@ -472,10 +472,10 @@ taos> select myfun(v1, v2) from t;
DB error: udf function execution failure (0.011088s)
```
不幸的是执行失败了，什么原因呢？查看 udfd 进程的日志。
不幸的是执行失败了，什么原因呢？查看 taosudf 进程的日志。
```shell
tail -10 /var/log/taos/udfd.log
tail -10 /var/log/taos/taosudf.log
```
发现以下错误信息。

18
include/libs/function/tudf.h
View File

@ -37,9 +37,9 @@ extern "C" {
#define UDF_LISTEN_PIPE_NAME_LEN 32
#ifdef _WIN32
#define UDF_LISTEN_PIPE_NAME_PREFIX "\\\\?\\pipe\\udfd.sock"
#define UDF_LISTEN_PIPE_NAME_PREFIX "\\\\?\\pipe\\taosudf.sock"
#else
#define UDF_LISTEN_PIPE_NAME_PREFIX ".udfd.sock."
#define UDF_LISTEN_PIPE_NAME_PREFIX ".taosudf.sock."
#endif
#define UDF_DNODE_ID_ENV_NAME "DNODE_ID"
@ -66,7 +66,7 @@ extern "C" {
const void *ptrs[] = {__VA_ARGS__}; \
for (int i = 0; i < sizeof(ptrs) / sizeof(ptrs[0]); ++i) { \
if (ptrs[i] == NULL) { \
fnError("udfd %dth parameter invalid, NULL PTR.line:%d", i, __LINE__); \
fnError("taosudf %dth parameter invalid, NULL PTR.line:%d", i, __LINE__); \
return TSDB_CODE_INVALID_PARA; \
} \
} \
@ -77,7 +77,7 @@ extern "C" {
const void *ptrs[] = {__VA_ARGS__}; \
for (int i = 0; i < sizeof(ptrs) / sizeof(ptrs[0]); ++i) { \
if (ptrs[i] == NULL) { \
fnError("udfd %dth parameter invalid, NULL PTR.line:%d", i, __LINE__); \
fnError("taosudf %dth parameter invalid, NULL PTR.line:%d", i, __LINE__); \
return; \
} \
} \
@ -137,31 +137,31 @@ int32_t cleanUpUdfs();
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// udf api
/**
* create udfd proxy, called once in process that call doSetupUdf/callUdfxxx/doTeardownUdf
* create taosudf proxy, called once in process that call doSetupUdf/callUdfxxx/doTeardownUdf
* @return error code
*/
int32_t udfcOpen();
/**
* destroy udfd proxy
* destroy taosudf proxy
* @return error code
*/
int32_t udfcClose();
/**
* start udfd that serves udf function invocation under dnode startDnodeId
* start taosudf that serves udf function invocation under dnode startDnodeId
* @param startDnodeId
* @return
*/
int32_t udfStartUdfd(int32_t startDnodeId);
/**
* stop udfd
* stop taosudf
* @return
*/
void udfStopUdfd();
/**
* get udfd pid
* get taosudf pid
*
*/
// int32_t udfGetUdfdPid(int32_t* pUdfdPid);

2
packaging/deb/DEBIAN/prerm
View File

@ -30,7 +30,7 @@ else
# Remove all links
${csudo}rm -f ${bin_link_dir}/taos || :
${csudo}rm -f ${bin_link_dir}/taosd || :
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/taosudf || :
${csudo}rm -f ${bin_link_dir}/taosadapter || :
${csudo}rm -f ${bin_link_dir}/taosdemo || :
${csudo}rm -f ${bin_link_dir}/taoskeeper || :

2
packaging/deb/makedeb.sh
View File

@ -103,7 +103,7 @@ sed -i "s/versionType=\"enterprise\"/versionType=\"community\"/g" ${pkg_dir}${in
cp ${compile_dir}/build/bin/taosd ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/udfd ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosudf ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosBenchmark ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosdump ${pkg_dir}${install_home_path}/bin

4
packaging/rpm/tdengine.spec
View File

@ -92,7 +92,7 @@ cp %{_compiledir}/../packaging/tools/set_core.sh %{buildroot}%{homepath}/bin
cp %{_compiledir}/../packaging/tools/taosd-dump-cfg.gdb %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taos %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosd %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/udfd %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosudf %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosBenchmark %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosdump %{buildroot}%{homepath}/bin
cp %{_compiledir}/../../enterprise/packaging/start-all.sh %{buildroot}%{homepath}/bin
@ -233,7 +233,7 @@ if [ $1 -eq 0 ];then
# Remove all links
${csudo}rm -f ${bin_link_dir}/taos || :
${csudo}rm -f ${bin_link_dir}/taosd || :
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/taosudf || :
${csudo}rm -f ${bin_link_dir}/taosadapter || :
${csudo}rm -f ${bin_link_dir}/taoskeeper || :
${csudo}rm -f ${bin_link_dir}/taosdump || :

4
packaging/tools/install.sh
View File

@ -19,7 +19,7 @@ script_dir=$(dirname $(readlink -f "$0"))
PREFIX="taos"
clientName="${PREFIX}"
serverName="${PREFIX}d"
udfdName="udfd"
udfdName="taosudf"
configFile="${PREFIX}.cfg"
productName="TDengine"
emailName="taosdata.com"
@ -156,7 +156,7 @@ done
#echo "verType=${verType} interactiveFqdn=${interactiveFqdn}"
tools=(${clientName} ${benchmarkName} ${dumpName} ${demoName} remove.sh udfd set_core.sh TDinsight.sh start_pre.sh start-all.sh stop-all.sh)
tools=(${clientName} ${benchmarkName} ${dumpName} ${demoName} remove.sh taosudf set_core.sh TDinsight.sh start_pre.sh start-all.sh stop-all.sh)
if [ "${verMode}" == "cluster" ]; then
services=(${serverName} ${adapterName} ${xname} ${explorerName} ${keeperName})
elif [ "${verMode}" == "edge" ]; then

2
packaging/tools/make_install.bat
View File

@ -114,7 +114,7 @@ if %Enterprise% == TRUE (
)
copy %binary_dir%\\build\\bin\\taosd.exe %target_dir% > nul
copy %binary_dir%\\build\\bin\\udfd.exe %target_dir% > nul
copy %binary_dir%\\build\\bin\\taosudf.exe %target_dir% > nul
if exist %binary_dir%\\build\\bin\\taosadapter.exe (
copy %binary_dir%\\build\\bin\\taosadapter.exe %target_dir% > nul
)

10
packaging/tools/make_install.sh
View File

@ -171,7 +171,7 @@ function install_bin() {
${csudo}rm -f ${bin_link_dir}/${serverName} || :
${csudo}rm -f ${bin_link_dir}/taosadapter || :
${csudo}rm -f ${bin_link_dir}/taoskeeper || :
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/taosudf || :
${csudo}rm -f ${bin_link_dir}/taosdemo || :
${csudo}rm -f ${bin_link_dir}/taosdump || :
${csudo}rm -f ${bin_link_dir}/${uninstallScript} || :
@ -186,7 +186,7 @@ function install_bin() {
[ -f ${binary_dir}/build/bin/taosdump ] && ${csudo}cp -r ${binary_dir}/build/bin/taosdump ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosadapter ] && ${csudo}cp -r ${binary_dir}/build/bin/taosadapter ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taoskeeper ] && ${csudo}cp -r ${binary_dir}/build/bin/taoskeeper ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/udfd ] && ${csudo}cp -r ${binary_dir}/build/bin/udfd ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosudf ] && ${csudo}cp -r ${binary_dir}/build/bin/taosudf ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosx ] && ${csudo}cp -r ${binary_dir}/build/bin/taosx ${install_main_dir}/bin || :
${csudo}cp -r ${binary_dir}/build/bin/${serverName} ${install_main_dir}/bin || :
@ -201,7 +201,7 @@ function install_bin() {
[ -x ${install_main_dir}/bin/${serverName} ] && ${csudo}ln -s ${install_main_dir}/bin/${serverName} ${bin_link_dir}/${serverName} > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosadapter ] && ${csudo}ln -s ${install_main_dir}/bin/taosadapter ${bin_link_dir}/taosadapter > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taoskeeper ] && ${csudo}ln -s ${install_main_dir}/bin/taoskeeper ${bin_link_dir}/taoskeeper > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/udfd ] && ${csudo}ln -s ${install_main_dir}/bin/udfd ${bin_link_dir}/udfd > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosudf ] && ${csudo}ln -s ${install_main_dir}/bin/taosudf ${bin_link_dir}/taosudf > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosdump ] && ${csudo}ln -s ${install_main_dir}/bin/taosdump ${bin_link_dir}/taosdump > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosdemo ] && ${csudo}ln -s ${install_main_dir}/bin/taosdemo ${bin_link_dir}/taosdemo > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosx ] && ${csudo}ln -s ${install_main_dir}/bin/taosx ${bin_link_dir}/taosx > /dev/null 2>&1 || :
@ -216,7 +216,7 @@ function install_bin() {
[ -f ${binary_dir}/build/bin/taosdump ] && ${csudo}cp -r ${binary_dir}/build/bin/taosdump ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosadapter ] && ${csudo}cp -r ${binary_dir}/build/bin/taosadapter ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taoskeeper ] && ${csudo}cp -r ${binary_dir}/build/bin/taoskeeper ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/udfd ] && ${csudo}cp -r ${binary_dir}/build/bin/udfd ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosudf ] && ${csudo}cp -r ${binary_dir}/build/bin/taosudf ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/taosx ] && ${csudo}cp -r ${binary_dir}/build/bin/taosx ${install_main_dir}/bin || :
[ -f ${binary_dir}/build/bin/*explorer ] && ${csudo}cp -r ${binary_dir}/build/bin/*explorer ${install_main_dir}/bin || :
${csudo}cp -r ${binary_dir}/build/bin/${serverName} ${install_main_dir}/bin || :
@ -228,7 +228,7 @@ function install_bin() {
[ -x ${install_main_dir}/bin/${serverName} ] && ${csudo}ln -s ${install_main_dir}/bin/${serverName} ${bin_link_dir}/${serverName} > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosadapter ] && ${csudo}ln -s ${install_main_dir}/bin/taosadapter ${bin_link_dir}/taosadapter > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taoskeeper ] && ${csudo}ln -s ${install_main_dir}/bin/taoskeeper ${bin_link_dir}/taoskeeper > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/udfd ] && ${csudo}ln -s ${install_main_dir}/bin/udfd ${bin_link_dir}/udfd > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosudf ] && ${csudo}ln -s ${install_main_dir}/bin/taosudf ${bin_link_dir}/taosudf > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosdump ] && ${csudo}ln -s ${install_main_dir}/bin/taosdump ${bin_link_dir}/taosdump > /dev/null 2>&1 || :
[ -f ${install_main_dir}/bin/taosBenchmark ] && ${csudo}ln -sf ${install_main_dir}/bin/taosBenchmark ${install_main_dir}/bin/taosdemo > /dev/null 2>&1 || :
[ -x ${install_main_dir}/bin/taosx ] && ${csudo}ln -s ${install_main_dir}/bin/taosx ${bin_link_dir}/taosx > /dev/null 2>&1 || :

2
packaging/tools/makepkg.sh
View File

@ -98,7 +98,7 @@ else
${taostools_bin_files} \
${build_dir}/bin/${clientName}adapter \
${build_dir}/bin/${clientName}keeper \
${build_dir}/bin/udfd \
${build_dir}/bin/taosudf \
${script_dir}/remove.sh \
${script_dir}/set_core.sh \
${script_dir}/startPre.sh \

6
packaging/tools/post.sh
View File

@ -233,7 +233,7 @@ function install_bin() {
log_print "start install bin from ${bin_dir} to ${bin_link_dir}"
${csudo}rm -f ${bin_link_dir}/taos || :
${csudo}rm -f ${bin_link_dir}/taosd || :
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/taosudf || :
${csudo}rm -f ${bin_link_dir}/taosadapter || :
${csudo}rm -f ${bin_link_dir}/taosBenchmark || :
${csudo}rm -f ${bin_link_dir}/taoskeeper || :
@ -255,8 +255,8 @@ function install_bin() {
if [ -x ${bin_dir}/taosd ]; then
${csudo}ln -s ${bin_dir}/taosd ${bin_link_dir}/taosd 2>>${install_log_path} || return 1
fi
if [ -x ${bin_dir}/udfd ]; then
${csudo}ln -s ${bin_dir}/udfd ${bin_link_dir}/udfd 2>>${install_log_path} || return 1
if [ -x ${bin_dir}/taosudf ]; then
${csudo}ln -s ${bin_dir}/taosudf ${bin_link_dir}/taosudf 2>>${install_log_path} || return 1
fi
if [ -x ${bin_dir}/taosadapter ]; then
${csudo}ln -s ${bin_dir}/taosadapter ${bin_link_dir}/taosadapter 2>>${install_log_path} || return 1

2
packaging/tools/remove.sh
View File

@ -61,7 +61,7 @@ if [ "${verMode}" == "cluster" ]; then
else
services=(${PREFIX}"d" ${PREFIX}"adapter" ${PREFIX}"keeper" ${PREFIX}"-explorer")
fi
tools=(${PREFIX} ${PREFIX}"Benchmark" ${PREFIX}"dump" ${PREFIX}"demo" udfd set_core.sh TDinsight.sh $uninstallScript start-all.sh stop-all.sh)
tools=(${PREFIX} ${PREFIX}"Benchmark" ${PREFIX}"dump" ${PREFIX}"demo" taosudf set_core.sh TDinsight.sh $uninstallScript start-all.sh stop-all.sh)
csudo=""
if command -v sudo >/dev/null; then

2
source/common/src/tglobal.c
View File

@ -319,7 +319,7 @@ int32_t tsS3MigrateIntervalSec = 60 * 60; // interval of s3migrate db in all vg
bool tsS3MigrateEnabled = 0;
int32_t tsGrantHBInterval = 60;
int32_t tsUptimeInterval = 300; // seconds
char tsUdfdResFuncs[512] = ""; // udfd resident funcs that teardown when udfd exits
char tsUdfdResFuncs[512] = ""; // taosudf resident funcs that teardown when taosudf exits
char tsUdfdLdLibPath[512] = "";
bool tsDisableStream = false;
int64_t tsStreamBufferSize = 128 * 1024 * 1024;

2
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
View File

@ -88,7 +88,7 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
}
if ((code = udfStartUdfd(pMgmt->pData->dnodeId)) != 0) {
dError("failed to start udfd since %s", tstrerror(code));
dError("failed to start taosudf since %s", tstrerror(code));
}
if ((code = taosAnalyticsInit()) != 0) {

8
source/libs/function/CMakeLists.txt
View File

@ -33,14 +33,14 @@ target_link_libraries(
PUBLIC uv_a
)
add_executable(udfd src/udfd.c)
add_executable(taosudf src/udfd.c)
if(${TD_DARWIN})
target_compile_options(udfd PRIVATE -Wno-error=deprecated-non-prototype)
target_compile_options(taosudf PRIVATE -Wno-error=deprecated-non-prototype)
endif()
target_include_directories(
udfd
taosudf
PUBLIC
"${TD_SOURCE_DIR}/include/libs/function"
"${TD_SOURCE_DIR}/contrib/libuv/include"
@ -52,7 +52,7 @@ target_include_directories(
)
target_link_libraries(
udfd
taosudf
PUBLIC uv_a
PRIVATE os util common nodes function
)

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

@ -63,25 +63,25 @@ static void udfWatchUdfd(void *args);
void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int32_t termSignal) {
TAOS_UDF_CHECK_PTR_RVOID(process);
fnInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
fnInfo("taosudf process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
SUdfdData *pData = process->data;
if(pData == NULL) {
fnError("udfd process data is NULL");
fnError("taosudf process data is NULL");
return;
}
if (exitStatus == 0 && termSignal == 0 || atomic_load_32(&pData->stopCalled)) {
fnInfo("udfd process exit due to SIGINT or dnode-mgmt called stop");
fnInfo("taosudf process exit due to SIGINT or dnode-mgmt called stop");
} else {
fnInfo("udfd process restart");
fnInfo("taosudf process restart");
int32_t code = udfSpawnUdfd(pData);
if (code != 0) {
fnError("udfd process restart failed with code:%d", code);
fnError("taosudf process restart failed with code:%d", code);
}
}
}
static int32_t udfSpawnUdfd(SUdfdData *pData) {
fnInfo("start to init udfd");
fnInfo("start to init taosudf");
TAOS_UDF_CHECK_PTR_RCODE(pData);
int32_t err = 0;
@ -106,12 +106,12 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
if (strlen(path) == 0) {
TAOS_STRCAT(path, "C:\\TDengine");
}
TAOS_STRCAT(path, "\\udfd.exe");
TAOS_STRCAT(path, "\\taosudf.exe");
#else
if (strlen(path) == 0) {
TAOS_STRCAT(path, "/usr/bin");
}
TAOS_STRCAT(path, "/udfd");
TAOS_STRCAT(path, "/taosudf");
#endif
char *argsUdfd[] = {path, "-c", configDir, NULL};
options.args = argsUdfd;
@ -158,9 +158,9 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
udfdPathLdLib[udfdLdLibPathLen] = ':';
tstrncpy(udfdPathLdLib + udfdLdLibPathLen + 1, pathTaosdLdLib, sizeof(udfdPathLdLib) - udfdLdLibPathLen - 1);
if (udfdLdLibPathLen + taosdLdLibPathLen < 1024) {
fnInfo("[UDFD]udfd LD_LIBRARY_PATH: %s", udfdPathLdLib);
fnInfo("[UDFD]taosudf LD_LIBRARY_PATH: %s", udfdPathLdLib);
} else {
fnError("[UDFD]can not set correct udfd LD_LIBRARY_PATH");
fnError("[UDFD]can not set correct taosudf LD_LIBRARY_PATH");
}
char ldLibPathEnvItem[1024 + 32] = {0};
snprintf(ldLibPathEnvItem, 1024 + 32, "%s=%s", "LD_LIBRARY_PATH", udfdPathLdLib);
@ -231,12 +231,12 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
pData->process.data = (void *)pData;
#ifdef WINDOWS
// End udfd.exe by Job.
// End taosudf.exe by Job.
if (pData->jobHandle != NULL) CloseHandle(pData->jobHandle);
pData->jobHandle = CreateJobObject(NULL, NULL);
bool add_job_ok = AssignProcessToJobObject(pData->jobHandle, pData->process.process_handle);
if (!add_job_ok) {
fnError("Assign udfd to job failed.");
fnError("Assign taosudf to job failed.");
} else {
JOBOBJECT_EXTENDED_LIMIT_INFORMATION limit_info;
memset(&limit_info, 0x0, sizeof(limit_info));
@ -244,15 +244,15 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
bool set_auto_kill_ok =
SetInformationJobObject(pData->jobHandle, JobObjectExtendedLimitInformation, &limit_info, sizeof(limit_info));
if (!set_auto_kill_ok) {
fnError("Set job auto kill udfd failed.");
fnError("Set job auto kill taosudf failed.");
}
}
#endif
if (err != 0) {
fnError("can not spawn udfd. path: %s, error: %s", path, uv_strerror(err));
fnError("can not spawn taosudf. path: %s, error: %s", path, uv_strerror(err));
} else {
fnInfo("udfd is initialized");
fnInfo("taosudf is initialized");
}
_OVER:
@ -295,13 +295,13 @@ static void udfWatchUdfd(void *args) {
atomic_store_32(&pData->spawnErr, 0);
(void)uv_barrier_wait(&pData->barrier);
int32_t num = uv_run(&pData->loop, UV_RUN_DEFAULT);
fnInfo("udfd loop exit with %d active handles, line:%d", num, __LINE__);
fnInfo("taosudf loop exit with %d active handles, line:%d", num, __LINE__);
uv_walk(&pData->loop, udfUdfdCloseWalkCb, NULL);
num = uv_run(&pData->loop, UV_RUN_DEFAULT);
fnInfo("udfd loop exit with %d active handles, line:%d", num, __LINE__);
fnInfo("taosudf loop exit with %d active handles, line:%d", num, __LINE__);
if (uv_loop_close(&pData->loop) != 0) {
fnError("udfd loop close failed, lino:%d", __LINE__);
fnError("taosudf loop close failed, lino:%d", __LINE__);
}
return;
@ -310,9 +310,9 @@ _exit:
(void)uv_barrier_wait(&pData->barrier);
atomic_store_32(&pData->spawnErr, terrno);
if (uv_loop_close(&pData->loop) != 0) {
fnError("udfd loop close failed, lino:%d", __LINE__);
fnError("taosudf loop close failed, lino:%d", __LINE__);
}
fnError("udfd thread exit with code:%d lino:%d", terrno, terrln);
fnError("taosudf thread exit with code:%d lino:%d", terrno, terrln);
terrno = TSDB_CODE_UDF_UV_EXEC_FAILURE;
}
return;
@ -321,11 +321,11 @@ _exit:
int32_t udfStartUdfd(int32_t startDnodeId) {
int32_t code = 0, lino = 0;
if (!tsStartUdfd) {
fnInfo("start udfd is disabled.") return 0;
fnInfo("start taosudf is disabled.") return 0;
}
SUdfdData *pData = &udfdGlobal;
if (pData->startCalled) {
fnInfo("dnode start udfd already called");
fnInfo("dnode start taosudf already called");
return 0;
}
pData->startCalled = true;
@ -341,27 +341,27 @@ int32_t udfStartUdfd(int32_t startDnodeId) {
if (err != 0) {
uv_barrier_destroy(&pData->barrier);
if (uv_async_send(&pData->stopAsync) != 0) {
fnError("start udfd: failed to send stop async");
fnError("start taosudf: failed to send stop async");
}
if (uv_thread_join(&pData->thread) != 0) {
fnError("start udfd: failed to join udfd thread");
fnError("start taosudf: failed to join taosudf thread");
}
pData->needCleanUp = false;
fnInfo("udfd is cleaned up after spawn err");
fnInfo("taosudf is cleaned up after spawn err");
TAOS_CHECK_GOTO(err, &lino, _exit);
} else {
pData->needCleanUp = true;
}
_exit:
if (code != 0) {
fnError("udfd start failed with code:%d, lino:%d", code, lino);
fnError("taosudf start failed with code:%d, lino:%d", code, lino);
}
return code;
}
void udfStopUdfd() {
SUdfdData *pData = &udfdGlobal;
fnInfo("udfd start to stop, need cleanup:%d, spawn err:%d", pData->needCleanUp, pData->spawnErr);
fnInfo("taosudf start to stop, need cleanup:%d, spawn err:%d", pData->needCleanUp, pData->spawnErr);
if (!pData->needCleanUp || atomic_load_32(&pData->stopCalled)) {
return;
}
@ -369,16 +369,16 @@ void udfStopUdfd() {
pData->needCleanUp = false;
uv_barrier_destroy(&pData->barrier);
if (uv_async_send(&pData->stopAsync) != 0) {
fnError("stop udfd: failed to send stop async");
fnError("stop taosudf: failed to send stop async");
}
if (uv_thread_join(&pData->thread) != 0) {
fnError("stop udfd: failed to join udfd thread");
fnError("stop taosudf: failed to join taosudf thread");
}
#ifdef WINDOWS
if (pData->jobHandle != NULL) CloseHandle(pData->jobHandle);
#endif
fnInfo("udfd is cleaned up");
fnInfo("taosudf is cleaned up");
return;
}
@ -2152,7 +2152,7 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
fnDebug("udfc call udf. callType: %d, funcHandle: %p", callType, handle);
SUdfcUvSession *session = (SUdfcUvSession *)handle;
if (session->udfUvPipe == NULL) {
fnError("No pipe to udfd");
fnError("No pipe to taosudf");
return TSDB_CODE_UDF_PIPE_NOT_EXIST;
}
SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
@ -2282,7 +2282,7 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) {
SUdfcUvSession *session = (SUdfcUvSession *)handle;
if (session->udfUvPipe == NULL) {
fnError("tear down udf. pipe to udfd does not exist. udf name: %s", session->udfName);
fnError("tear down udf. pipe to taosudf does not exist. udf name: %s", session->udfName);
taosMemoryFree(session);
return TSDB_CODE_UDF_PIPE_NOT_EXIST;
}

128
source/libs/function/src/udfd.c
View File

@ -166,7 +166,7 @@ int32_t udfdCPluginUdfScalarProc(SUdfDataBlock *block, SUdfColumn *resultCol, vo
if (ctx->scalarProcFunc) {
return ctx->scalarProcFunc(block, resultCol);
} else {
fnError("udfd c plugin scalar proc not implemented");
fnError("taosudf c plugin scalar proc not implemented");
return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
}
}
@ -177,7 +177,7 @@ int32_t udfdCPluginUdfAggStart(SUdfInterBuf *buf, void *udfCtx) {
if (ctx->aggStartFunc) {
return ctx->aggStartFunc(buf);
} else {
fnError("udfd c plugin aggregation start not implemented");
fnError("taosudf c plugin aggregation start not implemented");
return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
}
return 0;
@ -189,7 +189,7 @@ int32_t udfdCPluginUdfAggProc(SUdfDataBlock *block, SUdfInterBuf *interBuf, SUdf
if (ctx->aggProcFunc) {
return ctx->aggProcFunc(block, interBuf, newInterBuf);
} else {
fnError("udfd c plugin aggregation process not implemented");
fnError("taosudf c plugin aggregation process not implemented");
return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
}
}
@ -201,7 +201,7 @@ int32_t udfdCPluginUdfAggProc(SUdfDataBlock *block, SUdfInterBuf *interBuf, SUdf
// if (ctx->aggMergeFunc) {
// return ctx->aggMergeFunc(inputBuf1, inputBuf2, outputBuf);
// } else {
// fnError("udfd c plugin aggregation merge not implemented");
// fnError("taosudf c plugin aggregation merge not implemented");
// return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
// }
// }
@ -212,7 +212,7 @@ int32_t udfdCPluginUdfAggFinish(SUdfInterBuf *buf, SUdfInterBuf *resultData, voi
if (ctx->aggFinishFunc) {
return ctx->aggFinishFunc(buf, resultData);
} else {
fnError("udfd c plugin aggregation finish not implemented");
fnError("taosudf c plugin aggregation finish not implemented");
return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
}
return 0;
@ -667,7 +667,7 @@ void udfdFreeUdf(void *pData) {
if (pSudf->scriptPlugin != NULL) {
if(pSudf->scriptPlugin->udfDestroyFunc(pSudf->scriptUdfCtx) != 0) {
fnError("udfdFreeUdf: udfd destroy udf %s failed", pSudf->name);
fnError("udfdFreeUdf: taosudf destroy udf %s failed", pSudf->name);
}
}
@ -688,15 +688,15 @@ int32_t udfdGetOrCreateUdf(SUdf **ppUdf, const char *udfName) {
++(*pUdfHash)->refCount;
*ppUdf = *pUdfHash;
uv_mutex_unlock(&global.udfsMutex);
fnInfo("udfd reuse existing udf. udf %s udf version %d, udf created time %" PRIx64, (*ppUdf)->name, (*ppUdf)->version,
fnInfo("taosudf reuse existing udf. udf %s udf version %d, udf created time %" PRIx64, (*ppUdf)->name, (*ppUdf)->version,
(*ppUdf)->createdTime);
return 0;
} else {
(*pUdfHash)->expired = true;
fnInfo("udfd expired, check for new version. existing udf %s udf version %d, udf created time %" PRIx64,
fnInfo("taosudf expired, check for new version. existing udf %s udf version %d, udf created time %" PRIx64,
(*pUdfHash)->name, (*pUdfHash)->version, (*pUdfHash)->createdTime);
if(taosHashRemove(global.udfsHash, udfName, strlen(udfName)) != 0) {
fnError("udfdGetOrCreateUdf: udfd remove udf %s failed", udfName);
fnError("udfdGetOrCreateUdf: taosudf remove udf %s failed", udfName);
}
}
}
@ -1001,7 +1001,7 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
uv_cond_destroy(&udf->condReady);
uv_mutex_destroy(&udf->lock);
code = udf->scriptPlugin->udfDestroyFunc(udf->scriptUdfCtx);
fnDebug("udfd destroy function returns %d", code);
fnDebug("taosudf destroy function returns %d", code);
taosMemoryFree(udf);
}
@ -1063,7 +1063,7 @@ int32_t udfdSaveFuncBodyToFile(SFuncInfo *pFuncInfo, SUdf *udf) {
TAOS_UDF_CHECK_PTR_RCODE(pFuncInfo, udf);
if (!osDataSpaceAvailable()) {
terrno = TSDB_CODE_NO_DISKSPACE;
fnError("udfd create shared library failed since %s", terrstr());
fnError("taosudf create shared library failed since %s", terrstr());
return terrno;
}
@ -1072,22 +1072,22 @@ int32_t udfdSaveFuncBodyToFile(SFuncInfo *pFuncInfo, SUdf *udf) {
bool fileExist = !(taosStatFile(path, NULL, NULL, NULL) < 0);
if (fileExist) {
tstrncpy(udf->path, path, PATH_MAX);
fnInfo("udfd func body file. reuse existing file %s", path);
fnInfo("taosudf func body file. reuse existing file %s", path);
return TSDB_CODE_SUCCESS;
}
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (file == NULL) {
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(terrno));
fnError("taosudf write udf shared library: %s failed, error: %d %s", path, errno, strerror(terrno));
return TSDB_CODE_FILE_CORRUPTED;
}
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
if (count != pFuncInfo->codeSize) {
fnError("udfd write udf shared library failed");
fnError("taosudf write udf shared library failed");
return TSDB_CODE_FILE_CORRUPTED;
}
if(taosCloseFile(&file) != 0) {
fnError("udfdSaveFuncBodyToFile, udfd close file failed");
fnError("udfdSaveFuncBodyToFile, taosudf close file failed");
return TSDB_CODE_FILE_CORRUPTED;
}
@ -1106,7 +1106,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
}
if (pMsg->code != TSDB_CODE_SUCCESS) {
fnError("udfd rpc error, code:%s", tstrerror(pMsg->code));
fnError("taosudf rpc error, code:%s", tstrerror(pMsg->code));
msgInfo->code = pMsg->code;
goto _return;
}
@ -1114,7 +1114,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
if (msgInfo->rpcType == UDFD_RPC_MNODE_CONNECT) {
SConnectRsp connectRsp = {0};
if(tDeserializeSConnectRsp(pMsg->pCont, pMsg->contLen, &connectRsp) < 0){
fnError("udfd deserialize connect response failed");
fnError("taosudf deserialize connect response failed");
goto _return;
}
@ -1137,7 +1137,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
} else if (msgInfo->rpcType == UDFD_RPC_RETRIVE_FUNC) {
SRetrieveFuncRsp retrieveRsp = {0};
if(tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp) < 0){
fnError("udfd deserialize retrieve func response failed");
fnError("taosudf deserialize retrieve func response failed");
goto _return;
}
@ -1303,16 +1303,16 @@ int32_t udfdOpenClientRpc() {
}
void udfdCloseClientRpc() {
fnInfo("udfd begin closing rpc");
fnInfo("taosudf begin closing rpc");
rpcClose(global.clientRpc);
fnInfo("udfd finish closing rpc");
fnInfo("taosudf finish closing rpc");
}
void udfdOnWrite(uv_write_t *req, int status) {
TAOS_UDF_CHECK_PTR_RVOID(req);
SUvUdfWork *work = (SUvUdfWork *)req->data;
if (status < 0) {
fnError("udfd send response error, length:%zu code:%s", work->output.len, uv_err_name(status));
fnError("taosudf send response error, length:%zu code:%s", work->output.len, uv_err_name(status));
}
// remove work from the connection work list
if (work->conn != NULL) {
@ -1337,14 +1337,14 @@ void udfdSendResponse(uv_work_t *work, int status) {
if (udfWork->conn != NULL) {
uv_write_t *write_req = taosMemoryMalloc(sizeof(uv_write_t));
if(write_req == NULL) {
fnError("udfd send response error, malloc failed");
fnError("taosudf send response error, malloc failed");
taosMemoryFree(work);
return;
}
write_req->data = udfWork;
int32_t code = uv_write(write_req, udfWork->conn->client, &udfWork->output, 1, udfdOnWrite);
if (code != 0) {
fnError("udfd send response error %s", uv_strerror(code));
fnError("taosudf send response error %s", uv_strerror(code));
taosMemoryFree(write_req);
}
}
@ -1365,7 +1365,7 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
buf->base = ctx->inputBuf;
buf->len = ctx->inputCap;
} else {
fnError("udfd can not allocate enough memory") buf->base = NULL;
fnError("taosudf can not allocate enough memory") buf->base = NULL;
buf->len = 0;
}
} else if (ctx->inputTotal == -1 && ctx->inputLen < msgHeadSize) {
@ -1379,7 +1379,7 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
buf->base = ctx->inputBuf + ctx->inputLen;
buf->len = ctx->inputCap - ctx->inputLen;
} else {
fnError("udfd can not allocate enough memory") buf->base = NULL;
fnError("taosudf can not allocate enough memory") buf->base = NULL;
buf->len = 0;
}
}
@ -1387,7 +1387,7 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
if (pipe == NULL) {
fnError("udfd pipe is NULL, LINE:%d", __LINE__);
fnError("taosudf pipe is NULL, LINE:%d", __LINE__);
return false;
}
if (pipe->inputTotal == -1 && pipe->inputLen >= sizeof(int32_t)) {
@ -1407,12 +1407,12 @@ void udfdHandleRequest(SUdfdUvConn *conn) {
uv_work_t *work = taosMemoryMalloc(sizeof(uv_work_t));
if(work == NULL) {
fnError("udfd malloc work failed");
fnError("taosudf malloc work failed");
return;
}
SUvUdfWork *udfWork = taosMemoryMalloc(sizeof(SUvUdfWork));
if(udfWork == NULL) {
fnError("udfd malloc udf work failed");
fnError("taosudf malloc udf work failed");
taosMemoryFree(work);
return;
}
@ -1427,7 +1427,7 @@ void udfdHandleRequest(SUdfdUvConn *conn) {
work->data = udfWork;
if(uv_queue_work(global.loop, work, udfdProcessRequest, udfdSendResponse) != 0)
{
fnError("udfd queue work failed");
fnError("taosudf queue work failed");
taosMemoryFree(work);
taosMemoryFree(udfWork);
}
@ -1449,7 +1449,7 @@ void udfdPipeCloseCb(uv_handle_t *pipe) {
void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
TAOS_UDF_CHECK_PTR_RVOID(client, buf);
fnDebug("udfd read %zd bytes from client", nread);
fnDebug("taosudf read %zd bytes from client", nread);
if (nread == 0) return;
SUdfdUvConn *conn = client->data;
@ -1466,7 +1466,7 @@ void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
if (nread < 0) {
if (nread == UV_EOF) {
fnInfo("udfd pipe read EOF");
fnInfo("taosudf pipe read EOF");
} else {
fnError("Receive error %s", uv_err_name(nread));
}
@ -1477,26 +1477,26 @@ void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
void udfdOnNewConnection(uv_stream_t *server, int status) {
TAOS_UDF_CHECK_PTR_RVOID(server);
if (status < 0) {
fnError("udfd new connection error, code:%s", uv_strerror(status));
fnError("taosudf new connection error, code:%s", uv_strerror(status));
return;
}
int32_t code = 0;
uv_pipe_t *client = (uv_pipe_t *)taosMemoryMalloc(sizeof(uv_pipe_t));
if(client == NULL) {
fnError("udfd pipe malloc failed");
fnError("taosudf pipe malloc failed");
return;
}
code = uv_pipe_init(global.loop, client, 0);
if (code) {
fnError("udfd pipe init error %s", uv_strerror(code));
fnError("taosudf pipe init error %s", uv_strerror(code));
taosMemoryFree(client);
return;
}
if (uv_accept(server, (uv_stream_t *)client) == 0) {
SUdfdUvConn *ctx = taosMemoryMalloc(sizeof(SUdfdUvConn));
if(ctx == NULL) {
fnError("udfd conn malloc failed");
fnError("taosudf conn malloc failed");
goto _exit;
}
ctx->pWorkList = NULL;
@ -1508,7 +1508,7 @@ void udfdOnNewConnection(uv_stream_t *server, int status) {
ctx->client = (uv_stream_t *)client;
code = uv_read_start((uv_stream_t *)client, udfdAllocBuffer, udfdPipeRead);
if (code) {
fnError("udfd read start error %s", uv_strerror(code));
fnError("taosudf read start error %s", uv_strerror(code));
udfdUvHandleError(ctx);
taosMemoryFree(ctx);
taosMemoryFree(client);
@ -1522,7 +1522,7 @@ _exit:
void udfdIntrSignalHandler(uv_signal_t *handle, int signum) {
TAOS_UDF_CHECK_PTR_RVOID(handle);
fnInfo("udfd signal received: %d\n", signum);
fnInfo("taosudf signal received: %d\n", signum);
uv_fs_t req;
int32_t code = uv_fs_unlink(global.loop, &req, global.listenPipeName, NULL);
if(code) {
@ -1558,7 +1558,7 @@ static int32_t udfdParseArgs(int32_t argc, char *argv[]) {
}
static void udfdPrintVersion() {
(void)printf("udfd version: %s compatible_version: %s\n", td_version, td_compatible_version);
(void)printf("taosudf version: %s compatible_version: %s\n", td_version, td_compatible_version);
(void)printf("git: %s\n", td_gitinfo);
(void)printf("build: %s\n", td_buildinfo);
}
@ -1573,7 +1573,7 @@ void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *bu
TAOS_UDF_CHECK_PTR_RVOID(buf);
buf->base = taosMemoryMalloc(suggested_size);
if (buf->base == NULL) {
fnError("udfd ctrl pipe alloc buffer failed");
fnError("taosudf ctrl pipe alloc buffer failed");
return;
}
buf->len = suggested_size;
@ -1582,13 +1582,13 @@ void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *bu
void udfdCtrlReadCb(uv_stream_t *q, ssize_t nread, const uv_buf_t *buf) {
TAOS_UDF_CHECK_PTR_RVOID(q, buf);
if (nread < 0) {
fnError("udfd ctrl pipe read error. %s", uv_err_name(nread));
fnError("taosudf ctrl pipe read error. %s", uv_err_name(nread));
taosMemoryFree(buf->base);
uv_close((uv_handle_t *)q, NULL);
uv_stop(global.loop);
return;
}
fnError("udfd ctrl pipe read %zu bytes", nread);
fnError("taosudf ctrl pipe read %zu bytes", nread);
taosMemoryFree(buf->base);
}
@ -1604,7 +1604,7 @@ static void removeListeningPipe() {
static int32_t udfdUvInit() {
TAOS_CHECK_RETURN(uv_loop_init(global.loop));
if (tsStartUdfd) { // udfd is started by taosd, which shall exit when taosd exit
if (tsStartUdfd) { // taosudf is started by taosd, which shall exit when taosd exit
TAOS_CHECK_RETURN(uv_pipe_init(global.loop, &global.ctrlPipe, 1));
TAOS_CHECK_RETURN(uv_pipe_open(&global.ctrlPipe, 0));
TAOS_CHECK_RETURN(uv_read_start((uv_stream_t *)&global.ctrlPipe, udfdCtrlAllocBufCb, udfdCtrlReadCb));
@ -1642,13 +1642,13 @@ static void udfdCloseWalkCb(uv_handle_t *handle, void *arg) {
static int32_t udfdGlobalDataInit() {
uv_loop_t *loop = taosMemoryMalloc(sizeof(uv_loop_t));
if (loop == NULL) {
fnError("udfd init uv loop failed, mem overflow");
fnError("taosudf init uv loop failed, mem overflow");
return terrno;
}
global.loop = loop;
if (uv_mutex_init(&global.scriptPluginsMutex) != 0) {
fnError("udfd init script plugins mutex failed");
fnError("taosudf init script plugins mutex failed");
return TSDB_CODE_UDF_UV_EXEC_FAILURE;
}
@ -1659,7 +1659,7 @@ static int32_t udfdGlobalDataInit() {
// taosHashSetFreeFp(global.udfsHash, udfdFreeUdf);
if (uv_mutex_init(&global.udfsMutex) != 0) {
fnError("udfd init udfs mutex failed");
fnError("taosudf init udfs mutex failed");
return TSDB_CODE_UDF_UV_EXEC_FAILURE;
}
@ -1670,23 +1670,23 @@ static void udfdGlobalDataDeinit() {
uv_mutex_destroy(&global.udfsMutex);
uv_mutex_destroy(&global.scriptPluginsMutex);
taosMemoryFreeClear(global.loop);
fnInfo("udfd global data deinit");
fnInfo("taosudf global data deinit");
}
static void udfdRun() {
fnInfo("start udfd event loop");
fnInfo("start taosudf event loop");
int32_t code = uv_run(global.loop, UV_RUN_DEFAULT);
if(code != 0) {
fnError("udfd event loop still has active handles or requests.");
fnError("taosudf event loop still has active handles or requests.");
}
fnInfo("udfd event loop stopped.");
fnInfo("taosudf event loop stopped.");
(void)uv_loop_close(global.loop);
uv_walk(global.loop, udfdCloseWalkCb, NULL);
code = uv_run(global.loop, UV_RUN_DEFAULT);
if(code != 0) {
fnError("udfd event loop still has active handles or requests.");
fnError("taosudf event loop still has active handles or requests.");
}
(void)uv_loop_close(global.loop);
}
@ -1702,7 +1702,7 @@ int32_t udfdInitResidentFuncs() {
while ((token = strtok_r(pSave, ",", &pSave)) != NULL) {
char func[TSDB_FUNC_NAME_LEN + 1] = {0};
tstrncpy(func, token, TSDB_FUNC_NAME_LEN);
fnInfo("udfd add resident function %s", func);
fnInfo("taosudf add resident function %s", func);
if(taosArrayPush(global.residentFuncs, func) == NULL)
{
taosArrayDestroy(global.residentFuncs);
@ -1722,18 +1722,18 @@ void udfdDeinitResidentFuncs() {
int32_t code = 0;
if (udf->scriptPlugin->udfDestroyFunc) {
code = udf->scriptPlugin->udfDestroyFunc(udf->scriptUdfCtx);
fnDebug("udfd %s destroy function returns %d", funcName, code);
fnDebug("taosudf %s destroy function returns %d", funcName, code);
}
if(taosHashRemove(global.udfsHash, funcName, strlen(funcName)) != 0)
{
fnError("udfd remove resident function %s failed", funcName);
fnError("taosudf remove resident function %s failed", funcName);
}
taosMemoryFree(udf);
}
}
taosHashCleanup(global.udfsHash);
taosArrayDestroy(global.residentFuncs);
fnInfo("udfd resident functions are deinit");
fnInfo("taosudf resident functions are deinit");
}
int32_t udfdCreateUdfSourceDir() {
@ -1743,7 +1743,7 @@ int32_t udfdCreateUdfSourceDir() {
snprintf(global.udfDataDir, PATH_MAX, "%s/.udf", tsTempDir);
code = taosMkDir(global.udfDataDir);
}
fnInfo("udfd create udf source directory %s. result: %s", global.udfDataDir, tstrerror(code));
fnInfo("taosudf create udf source directory %s. result: %s", global.udfDataDir, tstrerror(code));
return code;
}
@ -1779,7 +1779,7 @@ int main(int argc, char *argv[]) {
if (udfdInitLog() != 0) {
// ignore create log failed, because this error no matter
(void)printf("failed to init udfd log.");
(void)printf("failed to init taosudf log.");
} else {
logInitialized = true; // log is initialized
}
@ -1790,20 +1790,20 @@ int main(int argc, char *argv[]) {
goto _exit;
}
cfgInitialized = true; // cfg is initialized
fnInfo("udfd start with config file %s", configDir);
fnInfo("taosudf start with config file %s", configDir);
if (initEpSetFromCfg(tsFirst, tsSecond, &global.mgmtEp) != 0) {
fnError("init ep set from cfg failed");
code = -3;
goto _exit;
}
fnInfo("udfd start with mnode ep %s", global.mgmtEp.epSet.eps[0].fqdn);
fnInfo("taosudf start with mnode ep %s", global.mgmtEp.epSet.eps[0].fqdn);
if (udfdOpenClientRpc() != 0) {
fnError("open rpc connection to mnode failed");
code = -4;
goto _exit;
}
fnInfo("udfd rpc client is opened");
fnInfo("taosudf rpc client is opened");
openClientRpcFinished = true; // rpc is opened
if (udfdCreateUdfSourceDir() != 0) {
@ -1812,7 +1812,7 @@ int main(int argc, char *argv[]) {
goto _exit;
}
udfSourceDirInited = true; // udf source dir is created
fnInfo("udfd udf source directory is created");
fnInfo("taosudf udf source directory is created");
if (udfdGlobalDataInit() != 0) {
fnError("init global data failed");
@ -1820,14 +1820,14 @@ int main(int argc, char *argv[]) {
goto _exit;
}
globalDataInited = true; // global data is inited
fnInfo("udfd global data is inited");
fnInfo("taosudf global data is inited");
if (udfdUvInit() != 0) {
fnError("uv init failure");
code = -7;
goto _exit;
}
fnInfo("udfd uv is inited");
fnInfo("taosudf uv is inited");
if (udfdInitResidentFuncs() != 0) {
fnError("init resident functions failed");
@ -1835,10 +1835,10 @@ int main(int argc, char *argv[]) {
goto _exit;
}
residentFuncsInited = true; // resident functions are inited
fnInfo("udfd resident functions are inited");
fnInfo("taosudf resident functions are inited");
udfdRun();
fnInfo("udfd exit normally");
fnInfo("taosudf exit normally");
removeListeningPipe();

22
tests/system-test/0-others/udfTest.py
View File

@ -622,17 +622,17 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
cfgPath = buildPath + "/../sim/dnode1/cfg"
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(3):
tdLog.info(" loop restart udfd %d_th" % i)
tdLog.info(" loop restart taosudf %d_th" % i)
tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep -v grep| grep -v defunct | awk '{print $2}'"
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -643,9 +643,9 @@ class TDTestCase:
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# # start udfd cmds
# # start taosudf cmds
# start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
# tdLog.info("start udfd : %s " % start_udfd)
# tdLog.info("start taosudf : %s " % start_udfd)
def test_function_name(self):
tdLog.info(" create function name is not build_in functions ")
@ -680,15 +680,15 @@ class TDTestCase:
time.sleep(2)
def test_udfd_cmd(self):
tdLog.info(" test udfd -V ")
os.system("udfd -V")
tdLog.info(" test udfd -c ")
os.system("udfd -c")
tdLog.info(" test taosudf -V ")
os.system("taosudf -V")
tdLog.info(" test taosudf -c ")
os.system("taosudf -c")
letters = string.ascii_letters + string.digits + '\\'
path = ''.join(random.choice(letters) for i in range(5000))
os.system(f"udfd -c {path}")
os.system(f"taosudf -c {path}")
def test_change_udf_normal(self, func_name):
# create function with normal file

12
tests/system-test/0-others/udf_cfg2.py
View File

@ -592,17 +592,17 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
cfgPath = buildPath + "/../sim/dnode1/cfg"
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(3):
tdLog.info(" loop restart udfd %d_th" % i)
tdLog.info(" loop restart taosudf %d_th" % i)
tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep -v grep| grep -v defunct | awk '{print $2}'"
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -613,9 +613,9 @@ class TDTestCase:
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# # start udfd cmds
# # start taosudf cmds
# start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
# tdLog.info("start udfd : %s " % start_udfd)
# tdLog.info("start taosudf : %s " % start_udfd)
def test_function_name(self):
tdLog.info(" create function name is not build_in functions ")

10
tests/system-test/0-others/udf_cluster.py
View File

@ -301,14 +301,14 @@ class TDTestCase:
cfgPath = dnode.cfgDir
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(5):
tdLog.info(" loop restart udfd %d_th at dnode_index : %s" % (i ,dnode.index))
tdLog.info(" loop restart taosudf %d_th at dnode_index : %s" % (i ,dnode.index))
self.basic_udf_query(dnode)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep %s | grep 'root' | grep -v grep| grep -v defunct | awk '{print $2}'"%cfgPath
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep %s | grep 'root' | grep -v grep| grep -v defunct | awk '{print $2}'"%cfgPath
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -317,7 +317,7 @@ class TDTestCase:
def test_restart_udfd_All_dnodes(self):
for dnode in self.TDDnodes.dnodes:
tdLog.info(" start restart udfd for dnode_index :%s" %dnode.index )
tdLog.info(" start restart taosudf for dnode_index :%s" %dnode.index )
self.restart_udfd(dnode)

14
tests/system-test/0-others/udf_create.py
View File

@ -570,7 +570,7 @@ class TDTestCase:
time.sleep(1)
hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
if hwnd:
os.system("TASKKILL /F /IM udfd.exe")
os.system("TASKKILL /F /IM taosudf.exe")
def unexpected_create(self):
if (platform.system().lower() == 'windows' and tdDnodes.dnodes[0].remoteIP == ""):
@ -627,17 +627,17 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
cfgPath = buildPath + "/../sim/dnode1/cfg"
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(3):
tdLog.info(" loop restart udfd %d_th" % i)
tdLog.info(" loop restart taosudf %d_th" % i)
tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep -v grep| grep -v defunct | awk '{print $2}'"
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -648,9 +648,9 @@ class TDTestCase:
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# # start udfd cmds
# # start taosudf cmds
# start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
# tdLog.info("start udfd : %s " % start_udfd)
# tdLog.info("start taosudf : %s " % start_udfd)
def test_function_name(self):
tdLog.info(" create function name is not build_in functions ")

12
tests/system-test/0-others/udf_restart_taosd.py
View File

@ -589,17 +589,17 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
cfgPath = buildPath + "/../sim/dnode1/cfg"
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(3):
tdLog.info(" loop restart udfd %d_th" % i)
tdLog.info(" loop restart taosudf %d_th" % i)
tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep -v grep| grep -v defunct | awk '{print $2}'"
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -610,9 +610,9 @@ class TDTestCase:
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# # start udfd cmds
# # start taosudf cmds
# start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
# tdLog.info("start udfd : %s " % start_udfd)
# tdLog.info("start taosudf : %s " % start_udfd)
def test_function_name(self):
tdLog.info(" create function name is not build_in functions ")

14
tests/test_new/udf/udf_create.py
View File

@ -570,7 +570,7 @@ class TDTestCase:
time.sleep(1)
hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
if hwnd:
os.system("TASKKILL /F /IM udfd.exe")
os.system("TASKKILL /F /IM taosudf.exe")
def unexpected_create(self):
if (platform.system().lower() == 'windows' and tdDnodes.dnodes[0].remoteIP == ""):
@ -627,17 +627,17 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
cfgPath = buildPath + "/../sim/dnode1/cfg"
udfdPath = buildPath +'/build/bin/udfd'
udfdPath = buildPath +'/build/bin/taosudf'
for i in range(3):
tdLog.info(" loop restart udfd %d_th" % i)
tdLog.info(" loop restart taosudf %d_th" % i)
tdSql.query("select udf2(sub1.c1 ,sub1.c2), udf2(sub2.c2 ,sub2.c1) from sub1, sub2 where sub1.ts=sub2.ts and sub1.c1 is not null")
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# stop udfd cmds
get_processID = "ps -ef | grep -w udfd | grep -v grep| grep -v defunct | awk '{print $2}'"
# stop taosudf cmds
get_processID = "ps -ef | grep -w taosudf | grep -v grep| grep -v defunct | awk '{print $2}'"
processID = subprocess.check_output(get_processID, shell=True).decode("utf-8")
stop_udfd = " kill -9 %s" % processID
os.system(stop_udfd)
@ -648,9 +648,9 @@ class TDTestCase:
tdSql.checkData(0,0,169.661427555)
tdSql.checkData(0,1,169.661427555)
# # start udfd cmds
# # start taosudf cmds
# start_udfd = "nohup " + udfdPath +'-c' +cfgPath +" > /dev/null 2>&1 &"
# tdLog.info("start udfd : %s " % start_udfd)
# tdLog.info("start taosudf : %s " % start_udfd)
def test_function_name(self):
tdLog.info(" create function name is not build_in functions ")