diff --git a/source/libs/function/inc/tudf.h b/source/libs/function/inc/tudf.h index cb1fe320d8..158f79667b 100644 --- a/source/libs/function/inc/tudf.h +++ b/source/libs/function/inc/tudf.h @@ -37,23 +37,24 @@ extern "C" { enum { UDFC_CODE_STOPPING = -1, UDFC_CODE_PIPE_READ_ERR = -2, - UDF_CODE_LOAD_UDF_FAILURE = -3, + UDFC_CODE_CONNECT_PIPE_ERR = -3, + UDFC_CODE_LOAD_UDF_FAILURE = -4, + UDFC_CODE_INVALID_STATE = -5 }; -typedef void *UdfcHandle; typedef void *UdfcFuncHandle; /** * create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf * @return error code */ -int32_t udfcOpen(UdfcHandle* proxyHandle); +int32_t udfcOpen(); /** * destroy udfd proxy * @return error code */ -int32_t udfcClose(UdfcHandle proxyhandle); +int32_t udfcClose(); /** @@ -62,7 +63,7 @@ int32_t udfcClose(UdfcHandle proxyhandle); * @param handle, out * @return error code */ -int32_t setupUdf(UdfcHandle proxyHandle, char udfName[], SEpSet *epSet, UdfcFuncHandle *handle); +int32_t setupUdf(char udfName[], UdfcFuncHandle *handle); typedef struct SUdfColumnMeta { int16_t type; diff --git a/source/libs/function/inc/tudfInt.h b/source/libs/function/inc/tudfInt.h index 2ea89de73c..4e7178f7fd 100644 --- a/source/libs/function/inc/tudfInt.h +++ b/source/libs/function/inc/tudfInt.h @@ -39,7 +39,6 @@ enum { typedef struct SUdfSetupRequest { char udfName[TSDB_FUNC_NAME_LEN]; - SEpSet epSet; } SUdfSetupRequest; typedef struct SUdfSetupResponse { diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index 907a19078e..f8a7e77814 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -137,11 +137,11 @@ typedef struct SUdfdProxy { int8_t gUdfcState; QUEUE gUdfTaskQueue; QUEUE gUvProcTaskQueue; - // int8_t gUdfcState = UDFC_STATE_INITAL; - // QUEUE gUdfTaskQueue = {0}; - // QUEUE gUvProcTaskQueue = {0}; + + int8_t initialized; } SUdfdProxy; +SUdfdProxy gUdfdProxy = {0}; typedef struct SUdfUvSession { SUdfdProxy *udfc; @@ -209,7 +209,6 @@ enum { UDFC_STATE_STARTNG, // starting after udfcOpen UDFC_STATE_READY, // started and begin to receive quests UDFC_STATE_STOPPING, // stopping after udfcClose - UDFC_STATUS_FINAL, // stopped }; int32_t getUdfdPipeName(char* pipeName, int32_t size) { @@ -226,13 +225,11 @@ int32_t getUdfdPipeName(char* pipeName, int32_t size) { int32_t encodeUdfSetupRequest(void **buf, const SUdfSetupRequest *setup) { int32_t len = 0; len += taosEncodeBinary(buf, setup->udfName, TSDB_FUNC_NAME_LEN); - len += taosEncodeSEpSet(buf, &setup->epSet); return len; } void* decodeUdfSetupRequest(const void* buf, SUdfSetupRequest *request) { buf = taosDecodeBinaryTo(buf, request->udfName, TSDB_FUNC_NAME_LEN); - buf = taosDecodeSEpSet((void*)buf, &request->epSet); return (void*)buf; } @@ -615,7 +612,7 @@ void onUdfcPipeClose(uv_handle_t *handle) { } int32_t udfcGetUvTaskResponseResult(SClientUdfTask *task, SClientUvTaskNode *uvTask) { - debugPrint("%s", "get uv task result"); + fnDebug("udfc get uv task result. task: %p", task); if (uvTask->type == UV_TASK_REQ_RSP) { if (uvTask->rspBuf.base != NULL) { SUdfResponse rsp; @@ -658,7 +655,6 @@ int32_t udfcGetUvTaskResponseResult(SClientUdfTask *task, SClientUvTaskNode *uvT } void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) { - debugPrint("%s", "client allocate buffer to receive from pipe"); SClientUvConn *conn = handle->data; SClientConnBuf *connBuf = &conn->readBuf; @@ -673,7 +669,7 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf buf->base = connBuf->buf; buf->len = connBuf->cap; } else { - //TODO: log error + fnError("udfc allocate buffer failure. size: %d", msgHeadSize); buf->base = NULL; buf->len = 0; } @@ -685,13 +681,13 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf buf->base = connBuf->buf + connBuf->len; buf->len = connBuf->cap - connBuf->len; } else { - //TODO: log error free connBuf->buf + fnError("udfc re-allocate buffer failure. size: %d", connBuf->cap); buf->base = NULL; buf->len = 0; } } - debugPrint("\tconn buf cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total); + fnTrace("conn buf cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total); } @@ -700,6 +696,7 @@ bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) { connBuf->total = *(int32_t *) (connBuf->buf); } if (connBuf->len == connBuf->cap && connBuf->total == connBuf->cap) { + fnTrace("udfc complete message is received, now handle it"); return true; } return false; @@ -707,10 +704,10 @@ bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) { void udfcUvHandleRsp(SClientUvConn *conn) { SClientConnBuf *connBuf = &conn->readBuf; - int64_t seqNum = *(int64_t *) (connBuf->buf + sizeof(int32_t)); // msglen int32_t then seqnum + int64_t seqNum = *(int64_t *) (connBuf->buf + sizeof(int32_t)); // msglen then seqnum if (QUEUE_EMPTY(&conn->taskQueue)) { - //LOG error + fnError("udfc no task waiting for response on connection"); return; } bool found = false; @@ -724,7 +721,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) { found = true; taskFound = task; } else { - //LOG error; + fnError("udfc more than one task waiting for the same response"); continue; } } @@ -738,7 +735,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) { uv_sem_post(&taskFound->taskSem); QUEUE_REMOVE(&taskFound->procTaskQueue); } else { - //TODO: LOG error + fnError("no task is waiting for the response."); } connBuf->buf = NULL; connBuf->total = -1; @@ -762,7 +759,7 @@ void udfcUvHandleError(SClientUvConn *conn) { } void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) { - debugPrint("%s, nread: %zd", "client read from pipe", nread); + fnTrace("udfc client %p, client read from pipe. nread: %zd", client, nread); if (nread == 0) return; SClientUvConn *conn = client->data; @@ -775,9 +772,9 @@ void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) { } if (nread < 0) { - debugPrint("\tclient read error: %s", uv_strerror(nread)); + fnError("udfc client pipe %p read error: %s", client, uv_strerror(nread)); if (nread == UV_EOF) { - //TODO: + fnError("udfc client pipe %p closed", client); } udfcUvHandleError(conn); } @@ -785,16 +782,15 @@ void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) { } void onUdfClientWrite(uv_write_t *write, int status) { - debugPrint("%s", "after writing to pipe"); SClientUvTaskNode *uvTask = write->data; + uv_pipe_t *pipe = uvTask->pipe; if (status == 0) { - uv_pipe_t *pipe = uvTask->pipe; SClientUvConn *conn = pipe->data; QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue); } else { - //TODO Log error; + fnError("udfc client %p write error.", pipe); } - debugPrint("\tlength:%zu", uvTask->reqBuf.len); + fnTrace("udfc client %p write length:%zu", pipe, uvTask->reqBuf.len); taosMemoryFree(write); taosMemoryFree(uvTask->reqBuf.base); } @@ -852,7 +848,7 @@ int32_t createUdfcUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN } int32_t queueUvUdfTask(SClientUvTaskNode *uvTask) { - debugPrint("%s, %d", "queue uv task", uvTask->type); + fnTrace("queue uv task to event loop, task: %d, %p", uvTask->type, uvTask); SUdfdProxy *udfc = uvTask->udfc; uv_mutex_lock(&udfc->gUdfTaskQueueMutex); QUEUE_INSERT_TAIL(&udfc->gUdfTaskQueue, &uvTask->recvTaskQueue); @@ -866,7 +862,7 @@ int32_t queueUvUdfTask(SClientUvTaskNode *uvTask) { } int32_t startUvUdfTask(SClientUvTaskNode *uvTask) { - debugPrint("%s, type %d", "start uv task ", uvTask->type); + fnTrace("event loop start uv task. task: %d, %p", uvTask->type, uvTask); switch (uvTask->type) { case UV_TASK_CONNECT: { uv_pipe_t *pipe = taosMemoryMalloc(sizeof(uv_pipe_t)); @@ -981,27 +977,37 @@ void constructUdfService(void *argsThread) { uv_loop_close(&udfc->gUdfdLoop); } -int32_t udfcOpen(UdfcHandle *udfc) { - SUdfdProxy *proxy = taosMemoryCalloc(1, sizeof(SUdfdProxy)); +int32_t udfcOpen() { + int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 0, 1); + if (old == 1) { + return 0; + } + SUdfdProxy *proxy = &gUdfdProxy; getUdfdPipeName(proxy->udfdPipeName, UDF_LISTEN_PIPE_NAME_LEN); proxy->gUdfcState = UDFC_STATE_STARTNG; uv_barrier_init(&proxy->gUdfInitBarrier, 2); uv_thread_create(&proxy->gUdfLoopThread, constructUdfService, proxy); - uv_barrier_wait(&proxy->gUdfInitBarrier); + atomic_store_8(&proxy->gUdfcState, UDFC_STATE_READY); proxy->gUdfcState = UDFC_STATE_READY; - *udfc = proxy; + uv_barrier_wait(&proxy->gUdfInitBarrier); + fnInfo("udfc initialized") return 0; } -int32_t udfcClose(UdfcHandle udfcHandle) { - SUdfdProxy *udfc = udfcHandle; +int32_t udfcClose() { + int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 1, 0); + if (old == 0) { + return 0; + } + + SUdfdProxy *udfc = &gUdfdProxy; udfc->gUdfcState = UDFC_STATE_STOPPING; uv_async_send(&udfc->gUdfLoopStopAsync); uv_thread_join(&udfc->gUdfLoopThread); uv_mutex_destroy(&udfc->gUdfTaskQueueMutex); uv_barrier_destroy(&udfc->gUdfInitBarrier); - udfc->gUdfcState = UDFC_STATUS_FINAL; - taosMemoryFree(udfc); + udfc->gUdfcState = UDFC_STATE_INITAL; + fnInfo("udfc cleaned up"); return 0; } @@ -1019,29 +1025,36 @@ int32_t udfcRunUvTask(SClientUdfTask *task, int8_t uvTaskType) { return task->errCode; } -int32_t setupUdf(UdfcHandle udfc, char udfName[], SEpSet *epSet, UdfcFuncHandle *funcHandle) { - debugPrint("%s", "client setup udf"); +int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) { + fnInfo("udfc setup udf. udfName: %s", udfName); + if (gUdfdProxy.gUdfcState != UDFC_STATE_READY) { + return UDFC_CODE_INVALID_STATE; + } SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask)); task->errCode = 0; task->session = taosMemoryMalloc(sizeof(SUdfUvSession)); - task->session->udfc = udfc; + task->session->udfc = &gUdfdProxy; task->type = UDF_TASK_SETUP; SUdfSetupRequest *req = &task->_setup.req; memcpy(req->udfName, udfName, TSDB_FUNC_NAME_LEN); - req->epSet = *epSet; int32_t errCode = udfcRunUvTask(task, UV_TASK_CONNECT); if (errCode != 0) { - //TODO: log error - return -1; + fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName); + return UDFC_CODE_CONNECT_PIPE_ERR; } udfcRunUvTask(task, UV_TASK_REQ_RSP); SUdfSetupResponse *rsp = &task->_setup.rsp; task->session->severHandle = rsp->udfHandle; - *funcHandle = task->session; + if (task->errCode != 0) { + fnError("failed to setup udf. err: %d", task->errCode) + } else { + fnInfo("sucessfully setup udf func handle. handle: %p", task->session); + *funcHandle = task->session; + } int32_t err = task->errCode; taosMemoryFree(task); return err; @@ -1049,7 +1062,7 @@ int32_t setupUdf(UdfcHandle udfc, char udfName[], SEpSet *epSet, UdfcFuncHandle int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdfInterBuf *state, SUdfInterBuf *state2, SSDataBlock* output, SUdfInterBuf *newState) { - debugPrint("%s", "client call udf"); + fnTrace("udfc call udf. callType: %d, funcHandle: %p", callType, handle); SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask)); task->errCode = 0; @@ -1087,35 +1100,37 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf udfcRunUvTask(task, UV_TASK_REQ_RSP); - SUdfCallResponse *rsp = &task->_call.rsp; - switch (callType) { - case TSDB_UDF_CALL_AGG_INIT: { - *newState = rsp->resultBuf; - break; - } - case TSDB_UDF_CALL_AGG_PROC: { - *newState = rsp->resultBuf; - break; - } - case TSDB_UDF_CALL_AGG_MERGE: { - *newState = rsp->resultBuf; - break; - } - case TSDB_UDF_CALL_AGG_FIN: { - *newState = rsp->resultBuf; - break; - } - case TSDB_UDF_CALL_SCALA_PROC: { - *output = rsp->resultData; - break; + if (task->errCode != 0) { + fnError("call udf failure. err: %d", task->errCode); + } else { + SUdfCallResponse *rsp = &task->_call.rsp; + switch (callType) { + case TSDB_UDF_CALL_AGG_INIT: { + *newState = rsp->resultBuf; + break; + } + case TSDB_UDF_CALL_AGG_PROC: { + *newState = rsp->resultBuf; + break; + } + case TSDB_UDF_CALL_AGG_MERGE: { + *newState = rsp->resultBuf; + break; + } + case TSDB_UDF_CALL_AGG_FIN: { + *newState = rsp->resultBuf; + break; + } + case TSDB_UDF_CALL_SCALA_PROC: { + *output = rsp->resultData; + break; + } } } - taosMemoryFree(task); return task->errCode; } -//TODO: translate these calls to callUdf int32_t callUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf) { int8_t callType = TSDB_UDF_CALL_AGG_INIT; @@ -1159,7 +1174,7 @@ int32_t callUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t nu } int32_t teardownUdf(UdfcFuncHandle handle) { - debugPrint("%s", "client teardown udf"); + fnInfo("tear down udf. udf func handle: %p", handle); SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask)); task->errCode = 0; @@ -1171,7 +1186,6 @@ int32_t teardownUdf(UdfcFuncHandle handle) { udfcRunUvTask(task, UV_TASK_REQ_RSP); - SUdfTeardownResponse *rsp = &task->_teardown.rsp; int32_t err = task->errCode; diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c index 220e1cd025..e4c4cb4893 100644 --- a/source/libs/function/src/udfd.c +++ b/source/libs/function/src/udfd.c @@ -20,6 +20,7 @@ #include "tudf.h" #include "tudfInt.h" +#include "tdatablock.h" #include "tdataformat.h" #include "tglobal.h" #include "tmsg.h" @@ -31,8 +32,9 @@ typedef struct SUdfdContext { uv_signal_t intrSignal; char listenPipeName[UDF_LISTEN_PIPE_NAME_LEN]; uv_pipe_t listeningPipe; - void *clientRpc; + void *clientRpc; + SCorEpSet mgmtEp; uv_mutex_t udfsMutex; SHashObj *udfsHash; @@ -83,17 +85,17 @@ typedef struct SUdfcFuncHandle { SUdf *udf; } SUdfcFuncHandle; -int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName, SUdf *udf); +int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf); -int32_t udfdLoadUdf(char *udfName, SEpSet *pEpSet, SUdf *udf) { +int32_t udfdLoadUdf(char *udfName, SUdf *udf) { strcpy(udf->name, udfName); - udfdFillUdfInfoFromMNode(global.clientRpc, pEpSet, udf->name, udf); + udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf); //strcpy(udf->path, "/home/slzhou/TDengine/debug/build/lib/libudf1.so"); int err = uv_dlopen(udf->path, &udf->lib); if (err != 0) { fnError("can not load library %s. error: %s", udf->path, uv_strerror(err)); - return UDF_CODE_LOAD_UDF_FAILURE; + return UDFC_CODE_LOAD_UDF_FAILURE; } // TODO: find all the functions char normalFuncName[TSDB_FUNC_NAME_LEN] = {0}; @@ -140,7 +142,7 @@ void udfdProcessRequest(uv_work_t *req) { uv_mutex_lock(&udf->lock); if (udf->state == UDF_STATE_INIT) { udf->state = UDF_STATE_LOADING; - udfdLoadUdf(setup->udfName, &setup->epSet, udf); + udfdLoadUdf(setup->udfName, udf); udf->state = UDF_STATE_READY; uv_cond_broadcast(&udf->condReady); uv_mutex_unlock(&udf->lock); @@ -398,7 +400,48 @@ void udfdIntrSignalHandler(uv_signal_t *handle, int signum) { void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { return; } -int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName, SUdf *udf) { +int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) { + pEpSet->version = 0; + + // init mnode ip set + SEpSet* mgmtEpSet = &(pEpSet->epSet); + mgmtEpSet->numOfEps = 0; + mgmtEpSet->inUse = 0; + + if (firstEp && firstEp[0] != 0) { + if (strlen(firstEp) >= TSDB_EP_LEN) { + terrno = TSDB_CODE_TSC_INVALID_FQDN; + return -1; + } + + int32_t code = taosGetFqdnPortFromEp(firstEp, &mgmtEpSet->eps[0]); + if (code != TSDB_CODE_SUCCESS) { + terrno = TSDB_CODE_TSC_INVALID_FQDN; + return terrno; + } + + mgmtEpSet->numOfEps++; + } + + if (secondEp && secondEp[0] != 0) { + if (strlen(secondEp) >= TSDB_EP_LEN) { + terrno = TSDB_CODE_TSC_INVALID_FQDN; + return -1; + } + + taosGetFqdnPortFromEp(secondEp, &mgmtEpSet->eps[mgmtEpSet->numOfEps]); + mgmtEpSet->numOfEps++; + } + + if (mgmtEpSet->numOfEps == 0) { + terrno = TSDB_CODE_TSC_INVALID_FQDN; + return -1; + } + + return 0; +} + +int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) { SRetrieveFuncReq retrieveReq = {0}; retrieveReq.numOfFuncs = 1; retrieveReq.pFuncNames = taosArrayInit(1, TSDB_FUNC_NAME_LEN); @@ -415,7 +458,7 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName, rpcMsg.msgType = TDMT_MND_RETRIEVE_FUNC; SRpcMsg rpcRsp = {0}; - rpcSendRecv(clientRpc, pEpSet, &rpcMsg, &rpcRsp); + rpcSendRecv(clientRpc, &global.mgmtEp.epSet, &rpcMsg, &rpcRsp); SRetrieveFuncRsp retrieveRsp = {0}; tDeserializeSRetrieveFuncRsp(rpcRsp.pCont, rpcRsp.contLen, &retrieveRsp); @@ -618,5 +661,6 @@ int main(int argc, char *argv[]) { return -1; } + initEpSetFromCfg(tsFirst, tsSecond, &global.mgmtEp); return udfdRun(); } diff --git a/source/libs/function/test/runUdf.c b/source/libs/function/test/runUdf.c index 80313d251b..8e5eac538e 100644 --- a/source/libs/function/test/runUdf.c +++ b/source/libs/function/test/runUdf.c @@ -1,65 +1,84 @@ #include #include #include - #include "uv.h" + +#include "fnLog.h" #include "os.h" -#include "tudf.h" #include "tdatablock.h" +#include "tglobal.h" +#include "tudf.h" + +static int32_t parseArgs(int32_t argc, char *argv[]) { + for (int32_t i = 1; i < argc; ++i) { + if (strcmp(argv[i], "-c") == 0) { + if (i < argc - 1) { + if (strlen(argv[++i]) >= PATH_MAX) { + printf("config file path overflow"); + return -1; + } + tstrncpy(configDir, argv[i], PATH_MAX); + } else { + printf("'-c' requires a parameter, default is %s\n", configDir); + return -1; + } + } + } + + return 0; +} + +static int32_t initLog() { + char logName[12] = {0}; + snprintf(logName, sizeof(logName), "%slog", "udfc"); + return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, 0); +} int main(int argc, char *argv[]) { - UdfcHandle udfc; - udfcOpen(&udfc); - uv_sleep(1000); - char path[256] = {0}; - size_t cwdSize = 256; - int err = uv_cwd(path, &cwdSize); - if (err != 0) { - fprintf(stderr, "err cwd: %s\n", uv_strerror(err)); - return err; + parseArgs(argc, argv); + initLog(); + if (taosInitCfg(configDir, NULL, NULL, NULL, 0) != 0) { + fnError("failed to start since read config error"); + return -1; + } + + udfcOpen(); + uv_sleep(1000); + + UdfcFuncHandle handle; + + setupUdf("udf1", &handle); + + SSDataBlock block = {0}; + SSDataBlock *pBlock = █ + pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData)); + pBlock->info.numOfCols = 1; + pBlock->info.rows = 4; + char data[16] = {0}; + char bitmap[4] = {0}; + for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { + SColumnInfoData colInfo = {0}; + colInfo.info.type = TSDB_DATA_TYPE_INT; + colInfo.info.bytes = sizeof(int32_t); + colInfo.info.colId = 1; + colInfo.pData = data; + colInfo.nullbitmap = bitmap; + for (int32_t j = 0; j < pBlock->info.rows; ++j) { + colDataAppendInt32(&colInfo, j, &j); } - fprintf(stdout, "current working directory:%s\n", path); - strcat(path, "/libudf1.so"); + taosArrayPush(pBlock->pDataBlock, &colInfo); + } - UdfcFuncHandle handle; - SEpSet epSet; - epSet.inUse = 0; - taosGetFqdnPortFromEp("localhost:7100", &epSet.eps[0]); - taosGetFqdnPortFromEp("localhost:7200", &epSet.eps[1]); - epSet.numOfEps = 2; - setupUdf(udfc, "udf1", &epSet, &handle); + SScalarParam input = {0}; + input.numOfRows = pBlock->info.rows; + input.columnData = taosArrayGet(pBlock->pDataBlock, 0); + SScalarParam output = {0}; + callUdfScalarFunc(handle, &input, 1, &output); - SSDataBlock block = {0}; - SSDataBlock* pBlock = █ - pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData)); - pBlock->info.numOfCols = 1; - pBlock->info.rows = 4; - char data[16] = {0}; - char bitmap[4] = {0}; - for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { - SColumnInfoData colInfo = {0}; - colInfo.info.type = TSDB_DATA_TYPE_INT; - colInfo.info.bytes = sizeof(int32_t); - colInfo.info.colId = 1; - colInfo.pData = data; - colInfo.nullbitmap = bitmap; - for (int32_t j = 0; j < pBlock->info.rows; ++j) { - colDataAppendInt32(&colInfo, j, &j); - } - taosArrayPush(pBlock->pDataBlock, &colInfo); - } - - SScalarParam input = {0}; - input.numOfRows = pBlock->info.rows; - input.columnData = taosArrayGet(pBlock->pDataBlock, 0); - SScalarParam output = {0}; - callUdfScalarFunc(handle, &input, 1 , &output); - - SColumnInfoData *col = output.columnData; - for (int32_t i = 0; i < output.numOfRows; ++i) { - fprintf(stderr, "%d\t%d\n" , i, *(int32_t*)(col->pData + i *sizeof(int32_t))); - } - teardownUdf(handle); - - udfcClose(udfc); + SColumnInfoData *col = output.columnData; + for (int32_t i = 0; i < output.numOfRows; ++i) { + fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t))); + } + teardownUdf(handle); + udfcClose(); }