tudf.c 60.2 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
#include "uv.h"
H
Hongze Cheng 已提交
16

17
#include "os.h"
H
Hongze Cheng 已提交
18 19

#include "builtinsimpl.h"
S
slzhou 已提交
20
#include "fnLog.h"
H
Hongze Cheng 已提交
21 22
#include "functionMgt.h"
#include "querynodes.h"
S
shenglian zhou 已提交
23
#include "tarray.h"
S
slzhou 已提交
24
#include "tdatablock.h"
H
Hongze Cheng 已提交
25 26 27
#include "tglobal.h"
#include "tudf.h"
#include "tudfInt.h"
28

29
typedef struct SUdfdData {
H
Hongze Cheng 已提交
30 31 32 33 34 35
  bool         startCalled;
  bool         needCleanUp;
  uv_loop_t    loop;
  uv_thread_t  thread;
  uv_barrier_t barrier;
  uv_process_t process;
36
#ifdef WINDOWS
H
Hongze Cheng 已提交
37
  HANDLE jobHandle;
38
#endif
H
Hongze Cheng 已提交
39 40 41 42
  int        spawnErr;
  uv_pipe_t  ctrlPipe;
  uv_async_t stopAsync;
  int32_t    stopCalled;
43

H
Hongze Cheng 已提交
44
  int32_t dnodeId;
45 46 47 48
} SUdfdData;

SUdfdData udfdGlobal = {0};

S
shenglian zhou 已提交
49 50 51
int32_t udfStartUdfd(int32_t startDnodeId);
int32_t udfStopUdfd();

52
static int32_t udfSpawnUdfd(SUdfdData *pData);
H
Hongze Cheng 已提交
53 54 55 56 57
void           udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal);
static int32_t udfSpawnUdfd(SUdfdData *pData);
static void    udfUdfdCloseWalkCb(uv_handle_t *handle, void *arg);
static void    udfUdfdStopAsyncCb(uv_async_t *async);
static void    udfWatchUdfd(void *args);
58 59 60 61 62 63 64 65 66 67 68 69

void udfUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) {
  fnInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
  SUdfdData *pData = process->data;
  if (exitStatus == 0 && termSignal == 0 || atomic_load_32(&pData->stopCalled)) {
    fnInfo("udfd process exit due to SIGINT or dnode-mgmt called stop");
  } else {
    fnInfo("udfd process restart");
    udfSpawnUdfd(pData);
  }
}

H
Hongze Cheng 已提交
70
static int32_t udfSpawnUdfd(SUdfdData *pData) {
S
Shengliang Guan 已提交
71
  fnInfo("start to init udfd");
72 73 74 75 76
  uv_process_options_t options = {0};

  char path[PATH_MAX] = {0};
  if (tsProcPath == NULL) {
    path[0] = '.';
H
Hongze Cheng 已提交
77
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
78 79
    GetModuleFileName(NULL, path, PATH_MAX);
    taosDirName(path);
H
Hongze Cheng 已提交
80
#elif defined(_TD_DARWIN_64)
wafwerar's avatar
wafwerar 已提交
81 82 83
    uint32_t pathSize = sizeof(path);
    _NSGetExecutablePath(path, &pathSize);
    taosDirName(path);
H
Hongze Cheng 已提交
84
#endif
85
  } else {
86
    strncpy(path, tsProcPath, PATH_MAX);
87 88 89
    taosDirName(path);
  }
#ifdef WINDOWS
H
Hongze Cheng 已提交
90
  if (strlen(path) == 0) {
wafwerar's avatar
wafwerar 已提交
91 92 93 94
    strcat(path, "udfd.exe");
  } else {
    strcat(path, "\\udfd.exe");
  }
95 96 97
#else
  strcat(path, "/udfd");
#endif
H
Hongze Cheng 已提交
98
  char *argsUdfd[] = {path, "-c", configDir, NULL};
99 100 101 102 103 104 105 106 107
  options.args = argsUdfd;
  options.file = path;

  options.exit_cb = udfUdfdExit;

  uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1);

  uv_stdio_container_t child_stdio[3];
  child_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
H
Hongze Cheng 已提交
108
  child_stdio[0].data.stream = (uv_stream_t *)&pData->ctrlPipe;
109 110 111 112 113 114 115 116 117 118 119 120 121
  child_stdio[1].flags = UV_IGNORE;
  child_stdio[2].flags = UV_INHERIT_FD;
  child_stdio[2].data.fd = 2;
  options.stdio_count = 3;
  options.stdio = child_stdio;

  options.flags = UV_PROCESS_DETACHED;

  char dnodeIdEnvItem[32] = {0};
  char thrdPoolSizeEnvItem[32] = {0};
  snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pData->dnodeId);
  float numCpuCores = 4;
  taosGetCpuCores(&numCpuCores);
H
Hongze Cheng 已提交
122 123
  snprintf(thrdPoolSizeEnvItem, 32, "%s=%d", "UV_THREADPOOL_SIZE", (int)numCpuCores * 2);
  char *envUdfd[] = {dnodeIdEnvItem, thrdPoolSizeEnvItem, NULL};
124 125 126
  options.env = envUdfd;

  int err = uv_spawn(&pData->loop, &pData->process, &options);
H
Hongze Cheng 已提交
127
  pData->process.data = (void *)pData;
128

129 130 131 132 133 134 135 136 137 138 139
#ifdef WINDOWS
  // End udfd.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.");
  } else {
    JOBOBJECT_EXTENDED_LIMIT_INFORMATION limit_info;
    memset(&limit_info, 0x0, sizeof(limit_info));
    limit_info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
H
Hongze Cheng 已提交
140 141
    bool set_auto_kill_ok =
        SetInformationJobObject(pData->jobHandle, JobObjectExtendedLimitInformation, &limit_info, sizeof(limit_info));
142 143 144 145 146 147
    if (!set_auto_kill_ok) {
      fnError("Set job auto kill udfd failed.");
    }
  }
#endif

148 149
  if (err != 0) {
    fnError("can not spawn udfd. path: %s, error: %s", path, uv_strerror(err));
S
Shengliang Guan 已提交
150 151
  } else {
    fnInfo("udfd is initialized");
152 153 154 155
  }
  return err;
}

H
Hongze Cheng 已提交
156
static void udfUdfdCloseWalkCb(uv_handle_t *handle, void *arg) {
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184
  if (!uv_is_closing(handle)) {
    uv_close(handle, NULL);
  }
}

static void udfUdfdStopAsyncCb(uv_async_t *async) {
  SUdfdData *pData = async->data;
  uv_stop(&pData->loop);
}

static void udfWatchUdfd(void *args) {
  SUdfdData *pData = args;
  uv_loop_init(&pData->loop);
  uv_async_init(&pData->loop, &pData->stopAsync, udfUdfdStopAsyncCb);
  pData->stopAsync.data = pData;
  int32_t err = udfSpawnUdfd(pData);
  atomic_store_32(&pData->spawnErr, err);
  uv_barrier_wait(&pData->barrier);
  uv_run(&pData->loop, UV_RUN_DEFAULT);
  uv_loop_close(&pData->loop);

  uv_walk(&pData->loop, udfUdfdCloseWalkCb, NULL);
  uv_run(&pData->loop, UV_RUN_DEFAULT);
  uv_loop_close(&pData->loop);
  return;
}

int32_t udfStartUdfd(int32_t startDnodeId) {
S
slzhou 已提交
185
  if (!tsStartUdfd) {
H
Hongze Cheng 已提交
186
    fnInfo("start udfd is disabled.") return 0;
S
slzhou 已提交
187
  }
188 189
  SUdfdData *pData = &udfdGlobal;
  if (pData->startCalled) {
S
Shengliang Guan 已提交
190
    fnInfo("dnode start udfd already called");
191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207
    return 0;
  }
  pData->startCalled = true;
  char dnodeId[8] = {0};
  snprintf(dnodeId, sizeof(dnodeId), "%d", startDnodeId);
  uv_os_setenv("DNODE_ID", dnodeId);
  pData->dnodeId = startDnodeId;

  uv_barrier_init(&pData->barrier, 2);
  uv_thread_create(&pData->thread, udfWatchUdfd, pData);
  uv_barrier_wait(&pData->barrier);
  int32_t err = atomic_load_32(&pData->spawnErr);
  if (err != 0) {
    uv_barrier_destroy(&pData->barrier);
    uv_async_send(&pData->stopAsync);
    uv_thread_join(&pData->thread);
    pData->needCleanUp = false;
S
Shengliang Guan 已提交
208
    fnInfo("udfd is cleaned up after spawn err");
209 210 211 212 213 214 215 216
  } else {
    pData->needCleanUp = true;
  }
  return err;
}

int32_t udfStopUdfd() {
  SUdfdData *pData = &udfdGlobal;
H
Hongze Cheng 已提交
217
  fnInfo("udfd start to stop, need cleanup:%d, spawn err:%d", pData->needCleanUp, pData->spawnErr);
218 219 220 221 222 223 224 225
  if (!pData->needCleanUp || atomic_load_32(&pData->stopCalled)) {
    return 0;
  }
  atomic_store_32(&pData->stopCalled, 1);
  pData->needCleanUp = false;
  uv_barrier_destroy(&pData->barrier);
  uv_async_send(&pData->stopAsync);
  uv_thread_join(&pData->thread);
226 227 228
#ifdef WINDOWS
  if (pData->jobHandle != NULL) CloseHandle(pData->jobHandle);
#endif
S
Shengliang Guan 已提交
229
  fnInfo("udfd is cleaned up");
230 231 232 233
  return 0;
}

//==============================================================================================
S
shenglian zhou 已提交
234 235 236 237
/* Copyright (c) 2013, Ben Noordhuis <info@bnoordhuis.nl>
 * The QUEUE is copied from queue.h under libuv
 * */

S
shenglian zhou 已提交
238 239 240
typedef void *QUEUE[2];

/* Private macros. */
H
Hongze Cheng 已提交
241 242 243 244
#define QUEUE_NEXT(q)      (*(QUEUE **)&((*(q))[0]))
#define QUEUE_PREV(q)      (*(QUEUE **)&((*(q))[1]))
#define QUEUE_PREV_NEXT(q) (QUEUE_NEXT(QUEUE_PREV(q)))
#define QUEUE_NEXT_PREV(q) (QUEUE_PREV(QUEUE_NEXT(q)))
S
shenglian zhou 已提交
245 246

/* Public macros. */
H
Hongze Cheng 已提交
247
#define QUEUE_DATA(ptr, type, field) ((type *)((char *)(ptr)-offsetof(type, field)))
S
shenglian zhou 已提交
248 249 250 251

/* Important note: mutating the list while QUEUE_FOREACH is
 * iterating over its elements results in undefined behavior.
 */
H
Hongze Cheng 已提交
252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314
#define QUEUE_FOREACH(q, h) for ((q) = QUEUE_NEXT(h); (q) != (h); (q) = QUEUE_NEXT(q))

#define QUEUE_EMPTY(q) ((const QUEUE *)(q) == (const QUEUE *)QUEUE_NEXT(q))

#define QUEUE_HEAD(q) (QUEUE_NEXT(q))

#define QUEUE_INIT(q)    \
  do {                   \
    QUEUE_NEXT(q) = (q); \
    QUEUE_PREV(q) = (q); \
  } while (0)

#define QUEUE_ADD(h, n)                 \
  do {                                  \
    QUEUE_PREV_NEXT(h) = QUEUE_NEXT(n); \
    QUEUE_NEXT_PREV(n) = QUEUE_PREV(h); \
    QUEUE_PREV(h) = QUEUE_PREV(n);      \
    QUEUE_PREV_NEXT(h) = (h);           \
  } while (0)

#define QUEUE_SPLIT(h, q, n)       \
  do {                             \
    QUEUE_PREV(n) = QUEUE_PREV(h); \
    QUEUE_PREV_NEXT(n) = (n);      \
    QUEUE_NEXT(n) = (q);           \
    QUEUE_PREV(h) = QUEUE_PREV(q); \
    QUEUE_PREV_NEXT(h) = (h);      \
    QUEUE_PREV(q) = (n);           \
  } while (0)

#define QUEUE_MOVE(h, n)        \
  do {                          \
    if (QUEUE_EMPTY(h))         \
      QUEUE_INIT(n);            \
    else {                      \
      QUEUE *q = QUEUE_HEAD(h); \
      QUEUE_SPLIT(h, q, n);     \
    }                           \
  } while (0)

#define QUEUE_INSERT_HEAD(h, q)    \
  do {                             \
    QUEUE_NEXT(q) = QUEUE_NEXT(h); \
    QUEUE_PREV(q) = (h);           \
    QUEUE_NEXT_PREV(q) = (q);      \
    QUEUE_NEXT(h) = (q);           \
  } while (0)

#define QUEUE_INSERT_TAIL(h, q)    \
  do {                             \
    QUEUE_NEXT(q) = (h);           \
    QUEUE_PREV(q) = QUEUE_PREV(h); \
    QUEUE_PREV_NEXT(q) = (q);      \
    QUEUE_PREV(h) = (q);           \
  } while (0)

#define QUEUE_REMOVE(q)                 \
  do {                                  \
    QUEUE_PREV_NEXT(q) = QUEUE_NEXT(q); \
    QUEUE_NEXT_PREV(q) = QUEUE_PREV(q); \
  } while (0)

enum { UV_TASK_CONNECT = 0, UV_TASK_REQ_RSP = 1, UV_TASK_DISCONNECT = 2 };
315

316
int64_t gUdfTaskSeqNum = 0;
317
typedef struct SUdfcFuncStub {
S
slzhou 已提交
318
  char           udfName[TSDB_FUNC_NAME_LEN + 1];
319
  UdfcFuncHandle handle;
H
Hongze Cheng 已提交
320 321
  int32_t        refCount;
  int64_t        lastRefTime;
322 323
} SUdfcFuncStub;

324
typedef struct SUdfcProxy {
H
Hongze Cheng 已提交
325
  char         udfdPipeName[PATH_MAX + UDF_LISTEN_PIPE_NAME_LEN + 2];
326
  uv_barrier_t initBarrier;
327

328 329 330
  uv_loop_t   uvLoop;
  uv_thread_t loopThread;
  uv_async_t  loopTaskAync;
331

332
  uv_async_t loopStopAsync;
333

334 335 336 337
  uv_mutex_t taskQueueMutex;
  int8_t     udfcState;
  QUEUE      taskQueue;
  QUEUE      uvProcTaskQueue;
338

339
  uv_mutex_t udfStubsMutex;
H
Hongze Cheng 已提交
340
  SArray    *udfStubs;  // SUdfcFuncStub
341

342
  int8_t initialized;
343
} SUdfcProxy;
344

345
SUdfcProxy gUdfdProxy = {0};
346

S
slzhou 已提交
347
typedef struct SUdfcUvSession {
348
  SUdfcProxy *udfc;
H
Hongze Cheng 已提交
349 350
  int64_t     severHandle;
  uv_pipe_t  *udfUvPipe;
S
shenglian zhou 已提交
351 352 353 354

  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
slzhou 已提交
355

S
slzhou 已提交
356
  char udfName[TSDB_FUNC_NAME_LEN + 1];
S
slzhou 已提交
357
} SUdfcUvSession;
358 359

typedef struct SClientUvTaskNode {
360
  SUdfcProxy *udfc;
H
Hongze Cheng 已提交
361 362
  int8_t      type;
  int         errCode;
363 364 365

  uv_pipe_t *pipe;

H
Hongze Cheng 已提交
366
  int64_t  seqNum;
367 368 369 370 371
  uv_buf_t reqBuf;

  uv_sem_t taskSem;
  uv_buf_t rspBuf;

S
shenglian zhou 已提交
372 373 374
  QUEUE recvTaskQueue;
  QUEUE procTaskQueue;
  QUEUE connTaskQueue;
375 376 377 378 379
} SClientUvTaskNode;

typedef struct SClientUdfTask {
  int8_t type;

S
slzhou 已提交
380
  SUdfcUvSession *session;
381 382 383 384 385

  int32_t errCode;

  union {
    struct {
H
Hongze Cheng 已提交
386
      SUdfSetupRequest  req;
387 388 389
      SUdfSetupResponse rsp;
    } _setup;
    struct {
H
Hongze Cheng 已提交
390
      SUdfCallRequest  req;
391 392 393
      SUdfCallResponse rsp;
    } _call;
    struct {
H
Hongze Cheng 已提交
394
      SUdfTeardownRequest  req;
395 396 397 398 399 400 401
      SUdfTeardownResponse rsp;
    } _teardown;
  };

} SClientUdfTask;

typedef struct SClientConnBuf {
H
Hongze Cheng 已提交
402
  char   *buf;
403 404 405 406 407 408
  int32_t len;
  int32_t cap;
  int32_t total;
} SClientConnBuf;

typedef struct SClientUvConn {
H
Hongze Cheng 已提交
409 410 411
  uv_pipe_t      *pipe;
  QUEUE           taskQueue;
  SClientConnBuf  readBuf;
S
slzhou 已提交
412
  SUdfcUvSession *session;
413 414
} SClientUvConn;

415
enum {
H
Hongze Cheng 已提交
416 417 418 419
  UDFC_STATE_INITAL = 0,  // initial state
  UDFC_STATE_STARTNG,     // starting after udfcOpen
  UDFC_STATE_READY,       // started and begin to receive quests
  UDFC_STATE_STOPPING,    // stopping after udfcClose
420
};
421

H
Hongze Cheng 已提交
422
int32_t getUdfdPipeName(char *pipeName, int32_t size);
S
shenglian zhou 已提交
423
int32_t encodeUdfSetupRequest(void **buf, const SUdfSetupRequest *setup);
H
Hongze Cheng 已提交
424 425 426
void   *decodeUdfSetupRequest(const void *buf, SUdfSetupRequest *request);
int32_t encodeUdfInterBuf(void **buf, const SUdfInterBuf *state);
void   *decodeUdfInterBuf(const void *buf, SUdfInterBuf *state);
S
shenglian zhou 已提交
427
int32_t encodeUdfCallRequest(void **buf, const SUdfCallRequest *call);
H
Hongze Cheng 已提交
428
void   *decodeUdfCallRequest(const void *buf, SUdfCallRequest *call);
S
shenglian zhou 已提交
429
int32_t encodeUdfTeardownRequest(void **buf, const SUdfTeardownRequest *teardown);
H
Hongze Cheng 已提交
430 431 432
void   *decodeUdfTeardownRequest(const void *buf, SUdfTeardownRequest *teardown);
int32_t encodeUdfRequest(void **buf, const SUdfRequest *request);
void   *decodeUdfRequest(const void *buf, SUdfRequest *request);
S
shenglian zhou 已提交
433
int32_t encodeUdfSetupResponse(void **buf, const SUdfSetupResponse *setupRsp);
H
Hongze Cheng 已提交
434
void   *decodeUdfSetupResponse(const void *buf, SUdfSetupResponse *setupRsp);
S
shenglian zhou 已提交
435
int32_t encodeUdfCallResponse(void **buf, const SUdfCallResponse *callRsp);
H
Hongze Cheng 已提交
436 437 438 439 440 441 442 443 444
void   *decodeUdfCallResponse(const void *buf, SUdfCallResponse *callRsp);
int32_t encodeUdfTeardownResponse(void **buf, const SUdfTeardownResponse *teardownRsp);
void   *decodeUdfTeardownResponse(const void *buf, SUdfTeardownResponse *teardownResponse);
int32_t encodeUdfResponse(void **buf, const SUdfResponse *rsp);
void   *decodeUdfResponse(const void *buf, SUdfResponse *rsp);
void    freeUdfColumnData(SUdfColumnData *data, SUdfColumnMeta *meta);
void    freeUdfColumn(SUdfColumn *col);
void    freeUdfDataDataBlock(SUdfDataBlock *block);
void    freeUdfInterBuf(SUdfInterBuf *buf);
S
shenglian zhou 已提交
445 446 447 448 449
int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlock);
int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block);
int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SSDataBlock *output);
int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output);

H
Hongze Cheng 已提交
450
int32_t getUdfdPipeName(char *pipeName, int32_t size) {
451
  char    dnodeId[8] = {0};
wafwerar's avatar
wafwerar 已提交
452
  size_t  dnodeIdSize = sizeof(dnodeId);
453 454
  int32_t err = uv_os_getenv(UDF_DNODE_ID_ENV_NAME, dnodeId, &dnodeIdSize);
  if (err != 0) {
S
Shengliang Guan 已提交
455
    fnError("failed to get dnodeId from env since %s", uv_err_name(err));
456 457
    dnodeId[0] = '1';
  }
458
#ifdef _WIN32
H
Hongze Cheng 已提交
459 460
  snprintf(pipeName, size, "%s.%x.%s", UDF_LISTEN_PIPE_NAME_PREFIX, MurmurHash3_32(tsDataDir, strlen(tsDataDir)),
           dnodeId);
461 462 463
#else
  snprintf(pipeName, size, "%s/%s%s", tsDataDir, UDF_LISTEN_PIPE_NAME_PREFIX, dnodeId);
#endif
S
Shengliang Guan 已提交
464
  fnInfo("get dnodeId:%s from env, pipe path:%s", dnodeId, pipeName);
465 466 467
  return 0;
}

468 469 470
int32_t encodeUdfSetupRequest(void **buf, const SUdfSetupRequest *setup) {
  int32_t len = 0;
  len += taosEncodeBinary(buf, setup->udfName, TSDB_FUNC_NAME_LEN);
S
shenglian zhou 已提交
471 472
  return len;
}
473

H
Hongze Cheng 已提交
474
void *decodeUdfSetupRequest(const void *buf, SUdfSetupRequest *request) {
475
  buf = taosDecodeBinaryTo(buf, request->udfName, TSDB_FUNC_NAME_LEN);
H
Hongze Cheng 已提交
476
  return (void *)buf;
S
shenglian zhou 已提交
477
}
478

H
Hongze Cheng 已提交
479
int32_t encodeUdfInterBuf(void **buf, const SUdfInterBuf *state) {
480
  int32_t len = 0;
481
  len += taosEncodeFixedI8(buf, state->numOfResult);
482 483
  len += taosEncodeFixedI32(buf, state->bufLen);
  len += taosEncodeBinary(buf, state->buf, state->bufLen);
S
shenglian zhou 已提交
484 485
  return len;
}
486

H
Hongze Cheng 已提交
487
void *decodeUdfInterBuf(const void *buf, SUdfInterBuf *state) {
488
  buf = taosDecodeFixedI8(buf, &state->numOfResult);
489
  buf = taosDecodeFixedI32(buf, &state->bufLen);
H
Hongze Cheng 已提交
490 491
  buf = taosDecodeBinary(buf, (void **)&state->buf, state->bufLen);
  return (void *)buf;
S
shenglian zhou 已提交
492 493
}

494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509
int32_t encodeUdfCallRequest(void **buf, const SUdfCallRequest *call) {
  int32_t len = 0;
  len += taosEncodeFixedI64(buf, call->udfHandle);
  len += taosEncodeFixedI8(buf, call->callType);
  if (call->callType == TSDB_UDF_CALL_SCALA_PROC) {
    len += tEncodeDataBlock(buf, &call->block);
  } else if (call->callType == TSDB_UDF_CALL_AGG_INIT) {
    len += taosEncodeFixedI8(buf, call->initFirst);
  } else if (call->callType == TSDB_UDF_CALL_AGG_PROC) {
    len += tEncodeDataBlock(buf, &call->block);
    len += encodeUdfInterBuf(buf, &call->interBuf);
  } else if (call->callType == TSDB_UDF_CALL_AGG_MERGE) {
    len += encodeUdfInterBuf(buf, &call->interBuf);
    len += encodeUdfInterBuf(buf, &call->interBuf2);
  } else if (call->callType == TSDB_UDF_CALL_AGG_FIN) {
    len += encodeUdfInterBuf(buf, &call->interBuf);
510
  }
511
  return len;
512 513
}

H
Hongze Cheng 已提交
514
void *decodeUdfCallRequest(const void *buf, SUdfCallRequest *call) {
515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534
  buf = taosDecodeFixedI64(buf, &call->udfHandle);
  buf = taosDecodeFixedI8(buf, &call->callType);
  switch (call->callType) {
    case TSDB_UDF_CALL_SCALA_PROC:
      buf = tDecodeDataBlock(buf, &call->block);
      break;
    case TSDB_UDF_CALL_AGG_INIT:
      buf = taosDecodeFixedI8(buf, &call->initFirst);
      break;
    case TSDB_UDF_CALL_AGG_PROC:
      buf = tDecodeDataBlock(buf, &call->block);
      buf = decodeUdfInterBuf(buf, &call->interBuf);
      break;
    case TSDB_UDF_CALL_AGG_MERGE:
      buf = decodeUdfInterBuf(buf, &call->interBuf);
      buf = decodeUdfInterBuf(buf, &call->interBuf2);
      break;
    case TSDB_UDF_CALL_AGG_FIN:
      buf = decodeUdfInterBuf(buf, &call->interBuf);
      break;
535
  }
H
Hongze Cheng 已提交
536
  return (void *)buf;
S
shenglian zhou 已提交
537 538
}

539 540 541 542
int32_t encodeUdfTeardownRequest(void **buf, const SUdfTeardownRequest *teardown) {
  int32_t len = 0;
  len += taosEncodeFixedI64(buf, teardown->udfHandle);
  return len;
S
shenglian zhou 已提交
543 544
}

H
Hongze Cheng 已提交
545
void *decodeUdfTeardownRequest(const void *buf, SUdfTeardownRequest *teardown) {
546
  buf = taosDecodeFixedI64(buf, &teardown->udfHandle);
H
Hongze Cheng 已提交
547
  return (void *)buf;
S
shenglian zhou 已提交
548 549
}

H
Hongze Cheng 已提交
550
int32_t encodeUdfRequest(void **buf, const SUdfRequest *request) {
551 552 553 554
  int32_t len = 0;
  if (buf == NULL) {
    len += sizeof(request->msgLen);
  } else {
H
Hongze Cheng 已提交
555
    *(int32_t *)(*buf) = request->msgLen;
556 557 558 559 560 561 562 563 564 565 566 567
    *buf = POINTER_SHIFT(*buf, sizeof(request->msgLen));
  }
  len += taosEncodeFixedI64(buf, request->seqNum);
  len += taosEncodeFixedI8(buf, request->type);
  if (request->type == UDF_TASK_SETUP) {
    len += encodeUdfSetupRequest(buf, &request->setup);
  } else if (request->type == UDF_TASK_CALL) {
    len += encodeUdfCallRequest(buf, &request->call);
  } else if (request->type == UDF_TASK_TEARDOWN) {
    len += encodeUdfTeardownRequest(buf, &request->teardown);
  }
  return len;
S
shenglian zhou 已提交
568 569
}

H
Hongze Cheng 已提交
570 571
void *decodeUdfRequest(const void *buf, SUdfRequest *request) {
  request->msgLen = *(int32_t *)(buf);
S
slzhou 已提交
572
  buf = POINTER_SHIFT(buf, sizeof(request->msgLen));
S
shenglian zhou 已提交
573

574 575
  buf = taosDecodeFixedI64(buf, &request->seqNum);
  buf = taosDecodeFixedI8(buf, &request->type);
S
shenglian zhou 已提交
576 577

  if (request->type == UDF_TASK_SETUP) {
578
    buf = decodeUdfSetupRequest(buf, &request->setup);
S
shenglian zhou 已提交
579
  } else if (request->type == UDF_TASK_CALL) {
580 581 582
    buf = decodeUdfCallRequest(buf, &request->call);
  } else if (request->type == UDF_TASK_TEARDOWN) {
    buf = decodeUdfTeardownRequest(buf, &request->teardown);
S
shenglian zhou 已提交
583
  }
H
Hongze Cheng 已提交
584
  return (void *)buf;
S
shenglian zhou 已提交
585
}
586

587 588 589
int32_t encodeUdfSetupResponse(void **buf, const SUdfSetupResponse *setupRsp) {
  int32_t len = 0;
  len += taosEncodeFixedI64(buf, setupRsp->udfHandle);
S
shenglian zhou 已提交
590 591 592
  len += taosEncodeFixedI8(buf, setupRsp->outputType);
  len += taosEncodeFixedI32(buf, setupRsp->outputLen);
  len += taosEncodeFixedI32(buf, setupRsp->bufSize);
593 594
  return len;
}
595

H
Hongze Cheng 已提交
596
void *decodeUdfSetupResponse(const void *buf, SUdfSetupResponse *setupRsp) {
597
  buf = taosDecodeFixedI64(buf, &setupRsp->udfHandle);
S
shenglian zhou 已提交
598 599 600
  buf = taosDecodeFixedI8(buf, &setupRsp->outputType);
  buf = taosDecodeFixedI32(buf, &setupRsp->outputLen);
  buf = taosDecodeFixedI32(buf, &setupRsp->bufSize);
H
Hongze Cheng 已提交
601
  return (void *)buf;
S
shenglian zhou 已提交
602
}
603

604 605 606 607 608 609
int32_t encodeUdfCallResponse(void **buf, const SUdfCallResponse *callRsp) {
  int32_t len = 0;
  len += taosEncodeFixedI8(buf, callRsp->callType);
  switch (callRsp->callType) {
    case TSDB_UDF_CALL_SCALA_PROC:
      len += tEncodeDataBlock(buf, &callRsp->resultData);
610
      break;
611
    case TSDB_UDF_CALL_AGG_INIT:
S
slzhou 已提交
612
      len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
613 614
      break;
    case TSDB_UDF_CALL_AGG_PROC:
S
slzhou 已提交
615
      len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
616 617
      break;
    case TSDB_UDF_CALL_AGG_MERGE:
S
slzhou 已提交
618
      len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
619 620
      break;
    case TSDB_UDF_CALL_AGG_FIN:
S
slzhou 已提交
621
      len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
622 623
      break;
  }
624
  return len;
S
shenglian zhou 已提交
625 626
}

H
Hongze Cheng 已提交
627
void *decodeUdfCallResponse(const void *buf, SUdfCallResponse *callRsp) {
628 629 630 631 632 633
  buf = taosDecodeFixedI8(buf, &callRsp->callType);
  switch (callRsp->callType) {
    case TSDB_UDF_CALL_SCALA_PROC:
      buf = tDecodeDataBlock(buf, &callRsp->resultData);
      break;
    case TSDB_UDF_CALL_AGG_INIT:
S
slzhou 已提交
634
      buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
635 636
      break;
    case TSDB_UDF_CALL_AGG_PROC:
S
slzhou 已提交
637
      buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
638 639
      break;
    case TSDB_UDF_CALL_AGG_MERGE:
S
slzhou 已提交
640
      buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
641 642
      break;
    case TSDB_UDF_CALL_AGG_FIN:
S
slzhou 已提交
643
      buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
644
      break;
S
shenglian zhou 已提交
645
  }
H
Hongze Cheng 已提交
646
  return (void *)buf;
S
shenglian zhou 已提交
647 648
}

H
Hongze Cheng 已提交
649
int32_t encodeUdfTeardownResponse(void **buf, const SUdfTeardownResponse *teardownRsp) { return 0; }
S
shenglian zhou 已提交
650

H
Hongze Cheng 已提交
651
void *decodeUdfTeardownResponse(const void *buf, SUdfTeardownResponse *teardownResponse) { return (void *)buf; }
S
shenglian zhou 已提交
652

H
Hongze Cheng 已提交
653
int32_t encodeUdfResponse(void **buf, const SUdfResponse *rsp) {
654 655 656 657
  int32_t len = 0;
  if (buf == NULL) {
    len += sizeof(rsp->msgLen);
  } else {
H
Hongze Cheng 已提交
658
    *(int32_t *)(*buf) = rsp->msgLen;
659
    *buf = POINTER_SHIFT(*buf, sizeof(rsp->msgLen));
S
shenglian zhou 已提交
660 661
  }

S
slzhou 已提交
662 663 664
  if (buf == NULL) {
    len += sizeof(rsp->seqNum);
  } else {
H
Hongze Cheng 已提交
665
    *(int64_t *)(*buf) = rsp->seqNum;
S
slzhou 已提交
666 667 668
    *buf = POINTER_SHIFT(*buf, sizeof(rsp->seqNum));
  }

669 670 671
  len += taosEncodeFixedI64(buf, rsp->seqNum);
  len += taosEncodeFixedI8(buf, rsp->type);
  len += taosEncodeFixedI32(buf, rsp->code);
S
shenglian zhou 已提交
672

673 674 675 676 677 678 679 680 681 682 683
  switch (rsp->type) {
    case UDF_TASK_SETUP:
      len += encodeUdfSetupResponse(buf, &rsp->setupRsp);
      break;
    case UDF_TASK_CALL:
      len += encodeUdfCallResponse(buf, &rsp->callRsp);
      break;
    case UDF_TASK_TEARDOWN:
      len += encodeUdfTeardownResponse(buf, &rsp->teardownRsp);
      break;
    default:
S
shenglian zhou 已提交
684
      fnError("encode udf response, invalid udf response type %d", rsp->type);
685 686 687
      break;
  }
  return len;
S
shenglian zhou 已提交
688 689
}

H
Hongze Cheng 已提交
690 691
void *decodeUdfResponse(const void *buf, SUdfResponse *rsp) {
  rsp->msgLen = *(int32_t *)(buf);
S
slzhou 已提交
692
  buf = POINTER_SHIFT(buf, sizeof(rsp->msgLen));
H
Hongze Cheng 已提交
693
  rsp->seqNum = *(int64_t *)(buf);
S
slzhou 已提交
694
  buf = POINTER_SHIFT(buf, sizeof(rsp->seqNum));
695 696 697
  buf = taosDecodeFixedI64(buf, &rsp->seqNum);
  buf = taosDecodeFixedI8(buf, &rsp->type);
  buf = taosDecodeFixedI32(buf, &rsp->code);
S
shenglian zhou 已提交
698

699 700 701 702 703 704 705 706 707 708 709
  switch (rsp->type) {
    case UDF_TASK_SETUP:
      buf = decodeUdfSetupResponse(buf, &rsp->setupRsp);
      break;
    case UDF_TASK_CALL:
      buf = decodeUdfCallResponse(buf, &rsp->callRsp);
      break;
    case UDF_TASK_TEARDOWN:
      buf = decodeUdfTeardownResponse(buf, &rsp->teardownRsp);
      break;
    default:
S
shenglian zhou 已提交
710
      fnError("decode udf response, invalid udf response type %d", rsp->type);
711
      break;
712
  }
H
Hongze Cheng 已提交
713
  return (void *)buf;
714
}
715

S
shenglian zhou 已提交
716 717
void freeUdfColumnData(SUdfColumnData *data, SUdfColumnMeta *meta) {
  if (IS_VAR_DATA_TYPE(meta->type)) {
S
slzhou 已提交
718 719 720 721
    taosMemoryFree(data->varLenCol.varOffsets);
    data->varLenCol.varOffsets = NULL;
    taosMemoryFree(data->varLenCol.payload);
    data->varLenCol.payload = NULL;
S
shenglian zhou 已提交
722
  } else {
S
slzhou 已提交
723 724 725 726
    taosMemoryFree(data->fixLenCol.nullBitmap);
    data->fixLenCol.nullBitmap = NULL;
    taosMemoryFree(data->fixLenCol.data);
    data->fixLenCol.data = NULL;
S
shenglian zhou 已提交
727 728 729
  }
}

H
Hongze Cheng 已提交
730
void freeUdfColumn(SUdfColumn *col) { freeUdfColumnData(&col->colData, &col->colMeta); }
S
shenglian zhou 已提交
731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746

void freeUdfDataDataBlock(SUdfDataBlock *block) {
  for (int32_t i = 0; i < block->numOfCols; ++i) {
    freeUdfColumn(block->udfCols[i]);
    taosMemoryFree(block->udfCols[i]);
    block->udfCols[i] = NULL;
  }
  taosMemoryFree(block->udfCols);
  block->udfCols = NULL;
}

void freeUdfInterBuf(SUdfInterBuf *buf) {
  taosMemoryFree(buf->buf);
  buf->buf = NULL;
}

S
slzhou 已提交
747 748
int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlock) {
  udfBlock->numOfRows = block->info.rows;
749
  udfBlock->numOfCols = taosArrayGetSize(block->pDataBlock);
H
Hongze Cheng 已提交
750
  udfBlock->udfCols = taosMemoryCalloc(taosArrayGetSize(block->pDataBlock), sizeof(SUdfColumn *));
S
slzhou 已提交
751
  for (int32_t i = 0; i < udfBlock->numOfCols; ++i) {
S
slzhou 已提交
752
    udfBlock->udfCols[i] = taosMemoryCalloc(1, sizeof(SUdfColumn));
H
Hongze Cheng 已提交
753 754
    SColumnInfoData *col = (SColumnInfoData *)taosArrayGet(block->pDataBlock, i);
    SUdfColumn      *udfCol = udfBlock->udfCols[i];
S
slzhou 已提交
755 756 757 758 759
    udfCol->colMeta.type = col->info.type;
    udfCol->colMeta.bytes = col->info.bytes;
    udfCol->colMeta.scale = col->info.scale;
    udfCol->colMeta.precision = col->info.precision;
    udfCol->colData.numOfRows = udfBlock->numOfRows;
S
slzhou@taodata.com 已提交
760
    udfCol->hasNull = col->hasNull;
S
shenglian zhou 已提交
761
    if (IS_VAR_DATA_TYPE(udfCol->colMeta.type)) {
S
slzhou 已提交
762 763 764 765 766 767
      udfCol->colData.varLenCol.varOffsetsLen = sizeof(int32_t) * udfBlock->numOfRows;
      udfCol->colData.varLenCol.varOffsets = taosMemoryMalloc(udfCol->colData.varLenCol.varOffsetsLen);
      memcpy(udfCol->colData.varLenCol.varOffsets, col->varmeta.offset, udfCol->colData.varLenCol.varOffsetsLen);
      udfCol->colData.varLenCol.payloadLen = colDataGetLength(col, udfBlock->numOfRows);
      udfCol->colData.varLenCol.payload = taosMemoryMalloc(udfCol->colData.varLenCol.payloadLen);
      memcpy(udfCol->colData.varLenCol.payload, col->pData, udfCol->colData.varLenCol.payloadLen);
S
slzhou 已提交
768
    } else {
S
slzhou 已提交
769 770 771
      udfCol->colData.fixLenCol.nullBitmapLen = BitmapLen(udfCol->colData.numOfRows);
      int32_t bitmapLen = udfCol->colData.fixLenCol.nullBitmapLen;
      udfCol->colData.fixLenCol.nullBitmap = taosMemoryMalloc(udfCol->colData.fixLenCol.nullBitmapLen);
H
Hongze Cheng 已提交
772
      char *bitmap = udfCol->colData.fixLenCol.nullBitmap;
S
slzhou 已提交
773 774 775 776
      memcpy(bitmap, col->nullbitmap, bitmapLen);
      udfCol->colData.fixLenCol.dataLen = colDataGetLength(col, udfBlock->numOfRows);
      int32_t dataLen = udfCol->colData.fixLenCol.dataLen;
      udfCol->colData.fixLenCol.data = taosMemoryMalloc(udfCol->colData.fixLenCol.dataLen);
H
Hongze Cheng 已提交
777
      char *data = udfCol->colData.fixLenCol.data;
S
slzhou 已提交
778
      memcpy(data, col->pData, dataLen);
S
slzhou 已提交
779 780 781 782 783 784 785
    }
  }
  return 0;
}

int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block) {
  block->info.rows = udfCol->colData.numOfRows;
S
shenglian zhou 已提交
786
  block->info.hasVarCol = IS_VAR_DATA_TYPE(udfCol->colMeta.type);
S
slzhou 已提交
787 788 789 790

  block->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
  taosArraySetSize(block->pDataBlock, 1);
  SColumnInfoData *col = taosArrayGet(block->pDataBlock, 0);
H
Hongze Cheng 已提交
791
  SUdfColumnMeta  *meta = &udfCol->colMeta;
S
slzhou 已提交
792 793 794 795
  col->info.precision = meta->precision;
  col->info.bytes = meta->bytes;
  col->info.scale = meta->scale;
  col->info.type = meta->type;
S
slzhou@taodata.com 已提交
796
  col->hasNull = udfCol->hasNull;
S
slzhou 已提交
797 798 799
  SUdfColumnData *data = &udfCol->colData;

  if (!IS_VAR_DATA_TYPE(meta->type)) {
S
slzhou 已提交
800 801 802 803
    col->nullbitmap = taosMemoryMalloc(data->fixLenCol.nullBitmapLen);
    memcpy(col->nullbitmap, data->fixLenCol.nullBitmap, data->fixLenCol.nullBitmapLen);
    col->pData = taosMemoryMalloc(data->fixLenCol.dataLen);
    memcpy(col->pData, data->fixLenCol.data, data->fixLenCol.dataLen);
S
slzhou 已提交
804
  } else {
S
slzhou 已提交
805 806 807 808
    col->varmeta.offset = taosMemoryMalloc(data->varLenCol.varOffsetsLen);
    memcpy(col->varmeta.offset, data->varLenCol.varOffsets, data->varLenCol.varOffsetsLen);
    col->pData = taosMemoryMalloc(data->varLenCol.payloadLen);
    memcpy(col->pData, data->varLenCol.payload, data->varLenCol.payloadLen);
S
slzhou 已提交
809 810 811 812
  }
  return 0;
}

S
slzhou 已提交
813 814 815
int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SSDataBlock *output) {
  output->info.rows = input->numOfRows;
  output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
816 817
  for (int32_t i = 0; i < numOfCols; ++i) {
    taosArrayPush(output->pDataBlock, (input + i)->columnData);
818

H
Hongze Cheng 已提交
819
    if (IS_VAR_DATA_TYPE((input + i)->columnData->info.type)) {
820 821
      output->info.hasVarCol = true;
    }
822
  }
S
slzhou 已提交
823 824 825 826
  return 0;
}

int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) {
827
  if (taosArrayGetSize(input->pDataBlock) != 1) {
S
slzhou 已提交
828 829 830 831
    fnError("scalar function only support one column");
    return -1;
  }
  output->numOfRows = input->info.rows;
S
slzhou 已提交
832 833

  output->columnData = taosMemoryMalloc(sizeof(SColumnInfoData));
H
Hongze Cheng 已提交
834 835
  memcpy(output->columnData, taosArrayGet(input->pDataBlock, 0), sizeof(SColumnInfoData));
  output->colAlloced = true;
S
slzhou 已提交
836

S
slzhou 已提交
837 838
  return 0;
}
S
slzhou 已提交
839

S
shenglian zhou 已提交
840
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
H
Hongze Cheng 已提交
841
// memory layout |---SUdfAggRes----|-----final result-----|---inter result----|
S
shenglian zhou 已提交
842 843 844
typedef struct SUdfAggRes {
  int8_t finalResNum;
  int8_t interResNum;
H
Hongze Cheng 已提交
845 846
  char  *finalResBuf;
  char  *interResBuf;
S
shenglian zhou 已提交
847
} SUdfAggRes;
H
Hongze Cheng 已提交
848
void    onUdfcPipeClose(uv_handle_t *handle);
S
shenglian zhou 已提交
849
int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask);
H
Hongze Cheng 已提交
850 851 852 853 854 855 856
void    udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf);
bool    isUdfcUvMsgComplete(SClientConnBuf *connBuf);
void    udfcUvHandleRsp(SClientUvConn *conn);
void    udfcUvHandleError(SClientUvConn *conn);
void    onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf);
void    onUdfcPipeWrite(uv_write_t *write, int status);
void    onUdfcPipeConnect(uv_connect_t *connect, int status);
S
shenglian zhou 已提交
857
int32_t udfcInitializeUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskNode *uvTask);
S
shenglian zhou 已提交
858 859
int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask);
int32_t udfcStartUvTask(SClientUvTaskNode *uvTask);
H
Hongze Cheng 已提交
860 861 862 863
void    udfcAsyncTaskCb(uv_async_t *async);
void    cleanUpUvTasks(SUdfcProxy *udfc);
void    udfStopAsyncCb(uv_async_t *async);
void    constructUdfService(void *argsThread);
S
shenglian zhou 已提交
864 865
int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType);
int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle);
H
Hongze Cheng 已提交
866
int     compareUdfcFuncSub(const void *elem1, const void *elem2);
S
shenglian zhou 已提交
867
int32_t doTeardownUdf(UdfcFuncHandle handle);
868

S
shenglian zhou 已提交
869
int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdfInterBuf *state, SUdfInterBuf *state2,
H
Hongze Cheng 已提交
870
                SSDataBlock *output, SUdfInterBuf *newState);
S
shenglian zhou 已提交
871 872
int32_t doCallUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf);
int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInterBuf *state, SUdfInterBuf *newState);
H
Hongze Cheng 已提交
873 874
int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2,
                          SUdfInterBuf *resultBuf);
S
shenglian zhou 已提交
875
int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData);
H
Hongze Cheng 已提交
876
int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output);
S
shenglian zhou 已提交
877 878 879 880 881
int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols, SScalarParam *output);

int32_t udfcOpen();
int32_t udfcClose();

H
Hongze Cheng 已提交
882 883
int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle);
void    releaseUdfFuncHandle(char *udfName);
S
shenglian zhou 已提交
884 885
int32_t cleanUpUdfs();

H
Hongze Cheng 已提交
886 887
bool    udfAggGetEnv(struct SFunctionNode *pFunc, SFuncExecEnv *pEnv);
bool    udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo *pResultCellInfo);
S
shenglian zhou 已提交
888
int32_t udfAggProcess(struct SqlFunctionCtx *pCtx);
H
Hongze Cheng 已提交
889
int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock);
890

H
Hongze Cheng 已提交
891
int compareUdfcFuncSub(const void *elem1, const void *elem2) {
S
shenglian zhou 已提交
892 893 894
  SUdfcFuncStub *stub1 = (SUdfcFuncStub *)elem1;
  SUdfcFuncStub *stub2 = (SUdfcFuncStub *)elem2;
  return strcmp(stub1->udfName, stub2->udfName);
895 896
}

H
Hongze Cheng 已提交
897
int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) {
S
shenglian zhou 已提交
898 899 900
  int32_t code = 0;
  uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
  SUdfcFuncStub key = {0};
S
slzhou 已提交
901
  strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
S
shenglian zhou 已提交
902 903 904 905
  int32_t stubIndex = taosArraySearchIdx(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
  if (stubIndex != -1) {
    SUdfcFuncStub *foundStub = taosArrayGet(gUdfdProxy.udfStubs, stubIndex);
    UdfcFuncHandle handle = foundStub->handle;
H
Hongze Cheng 已提交
906
    if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) {
S
shenglian zhou 已提交
907 908 909 910 911
      *pHandle = foundStub->handle;
      ++foundStub->refCount;
      foundStub->lastRefTime = taosGetTimestampUs();
      uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
      return 0;
912
    } else {
H
Hongze Cheng 已提交
913 914
      fnInfo("invalid handle for %s, refCount: %d, last ref time: %" PRId64 ". remove it from cache", udfName,
             foundStub->refCount, foundStub->lastRefTime);
S
shenglian zhou 已提交
915
      taosArrayRemove(gUdfdProxy.udfStubs, stubIndex);
916
    }
S
shenglian zhou 已提交
917 918 919 920 921 922 923 924 925 926 927
  }
  *pHandle = NULL;
  code = doSetupUdf(udfName, pHandle);
  if (code == TSDB_CODE_SUCCESS) {
    SUdfcFuncStub stub = {0};
    strcpy(stub.udfName, udfName);
    stub.handle = *pHandle;
    ++stub.refCount;
    stub.lastRefTime = taosGetTimestampUs();
    taosArrayPush(gUdfdProxy.udfStubs, &stub);
    taosArraySort(gUdfdProxy.udfStubs, compareUdfcFuncSub);
928
  } else {
S
shenglian zhou 已提交
929
    *pHandle = NULL;
930 931
  }

S
shenglian zhou 已提交
932 933
  uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
  return code;
934 935
}

H
Hongze Cheng 已提交
936
void releaseUdfFuncHandle(char *udfName) {
S
shenglian zhou 已提交
937 938
  uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
  SUdfcFuncStub key = {0};
S
slzhou 已提交
939
  strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
S
shenglian zhou 已提交
940 941
  SUdfcFuncStub *foundStub = taosArraySearch(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
  if (!foundStub) {
942
    uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
S
shenglian zhou 已提交
943
    return;
944
  }
S
shenglian zhou 已提交
945 946
  if (foundStub->refCount > 0) {
    --foundStub->refCount;
947
  }
S
shenglian zhou 已提交
948
  uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
949 950
}

S
shenglian zhou 已提交
951
int32_t cleanUpUdfs() {
952 953 954 955 956
  int8_t initialized = atomic_load_8(&gUdfdProxy.initialized);
  if (!initialized) {
    return TSDB_CODE_SUCCESS;
  }

S
shenglian zhou 已提交
957
  uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
H
Hongze Cheng 已提交
958
  if (gUdfdProxy.udfStubs == NULL || taosArrayGetSize(gUdfdProxy.udfStubs) == 0) {
959 960 961
    uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
    return TSDB_CODE_SUCCESS;
  }
H
Hongze Cheng 已提交
962
  SArray *udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub));
963
  int32_t i = 0;
S
shenglian zhou 已提交
964 965 966 967 968 969
  while (i < taosArrayGetSize(gUdfdProxy.udfStubs)) {
    SUdfcFuncStub *stub = taosArrayGet(gUdfdProxy.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);
    } else {
H
Hongze Cheng 已提交
970 971
      fnInfo("udf still in use. udf name: %s, ref count: %d, last ref time: %" PRId64 ", handle: %p", stub->udfName,
             stub->refCount, stub->lastRefTime, stub->handle);
S
shenglian zhou 已提交
972
      UdfcFuncHandle handle = stub->handle;
H
Hongze Cheng 已提交
973
      if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) {
S
shenglian zhou 已提交
974
        taosArrayPush(udfStubs, stub);
975
      } else {
H
Hongze Cheng 已提交
976
        fnInfo("udf invalid handle for %s, refCount: %d, last ref time: %" PRId64 ". remove it from cache",
S
shenglian zhou 已提交
977
               stub->udfName, stub->refCount, stub->lastRefTime);
978
      }
979
    }
S
shenglian zhou 已提交
980
    ++i;
981
  }
S
shenglian zhou 已提交
982 983 984 985 986
  taosArrayDestroy(gUdfdProxy.udfStubs);
  gUdfdProxy.udfStubs = udfStubs;
  uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
  return 0;
}
987

S
shenglian zhou 已提交
988 989
int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols, SScalarParam *output) {
  UdfcFuncHandle handle = NULL;
H
Hongze Cheng 已提交
990
  int32_t        code = acquireUdfFuncHandle(udfName, &handle);
S
shenglian zhou 已提交
991 992 993 994 995 996 997 998
  if (code != 0) {
    return code;
  }
  SUdfcUvSession *session = handle;
  code = doCallUdfScalarFunc(handle, input, numOfCols, output);
  if (output->columnData == NULL) {
    fnError("udfc scalar function calculate error. no column data");
    code = TSDB_CODE_UDF_INVALID_OUTPUT_TYPE;
999
  } else {
S
shenglian zhou 已提交
1000
    if (session->outputType != output->columnData->info.type || session->outputLen != output->columnData->info.bytes) {
H
Hongze Cheng 已提交
1001 1002
      fnError("udfc scalar function calculate error. type mismatch. session type: %d(%d), output type: %d(%d)",
              session->outputType, session->outputLen, output->columnData->info.type, output->columnData->info.bytes);
S
shenglian zhou 已提交
1003 1004
      code = TSDB_CODE_UDF_INVALID_OUTPUT_TYPE;
    }
1005
  }
S
shenglian zhou 已提交
1006 1007
  releaseUdfFuncHandle(udfName);
  return code;
1008
}
1009

H
Hongze Cheng 已提交
1010
bool udfAggGetEnv(struct SFunctionNode *pFunc, SFuncExecEnv *pEnv) {
S
shenglian zhou 已提交
1011 1012
  if (fmIsScalarFunc(pFunc->funcId)) {
    return false;
S
shenglian zhou 已提交
1013
  }
S
shenglian zhou 已提交
1014 1015
  pEnv->calcMemSize = sizeof(SUdfAggRes) + pFunc->node.resType.bytes + pFunc->udfBufSize;
  return true;
1016 1017
}

H
Hongze Cheng 已提交
1018
bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo *pResultCellInfo) {
S
shenglian zhou 已提交
1019 1020 1021 1022
  if (functionSetup(pCtx, pResultCellInfo) != true) {
    return false;
  }
  UdfcFuncHandle handle;
H
Hongze Cheng 已提交
1023
  int32_t        udfCode = 0;
S
shenglian zhou 已提交
1024 1025 1026 1027 1028
  if ((udfCode = acquireUdfFuncHandle((char *)pCtx->udfName, &handle)) != 0) {
    fnError("udfAggInit error. step doSetupUdf. udf code: %d", udfCode);
    return false;
  }
  SUdfcUvSession *session = (SUdfcUvSession *)handle;
H
Hongze Cheng 已提交
1029 1030
  SUdfAggRes     *udfRes = (SUdfAggRes *)GET_ROWCELL_INTERBUF(pResultCellInfo);
  int32_t         envSize = sizeof(SUdfAggRes) + session->outputLen + session->bufSize;
S
shenglian zhou 已提交
1031 1032
  memset(udfRes, 0, envSize);

H
Hongze Cheng 已提交
1033 1034
  udfRes->finalResBuf = (char *)udfRes + sizeof(SUdfAggRes);
  udfRes->interResBuf = (char *)udfRes + sizeof(SUdfAggRes) + session->outputLen;
S
shenglian zhou 已提交
1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055

  SUdfInterBuf buf = {0};
  if ((udfCode = doCallUdfAggInit(handle, &buf)) != 0) {
    fnError("udfAggInit error. step doCallUdfAggInit. udf code: %d", udfCode);
    releaseUdfFuncHandle(pCtx->udfName);
    return false;
  }
  udfRes->interResNum = buf.numOfResult;
  if (buf.bufLen <= session->bufSize) {
    memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
  } else {
    fnError("udfc inter buf size %d is greater than function bufSize %d", buf.bufLen, session->bufSize);
    releaseUdfFuncHandle(pCtx->udfName);
    return false;
  }
  releaseUdfFuncHandle(pCtx->udfName);
  freeUdfInterBuf(&buf);
  return true;
}

int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
H
Hongze Cheng 已提交
1056
  int32_t        udfCode = 0;
S
shenglian zhou 已提交
1057 1058 1059 1060 1061 1062 1063
  UdfcFuncHandle handle = 0;
  if ((udfCode = acquireUdfFuncHandle((char *)pCtx->udfName, &handle)) != 0) {
    fnError("udfAggProcess  error. step acquireUdfFuncHandle. udf code: %d", udfCode);
    return udfCode;
  }

  SUdfcUvSession *session = handle;
H
Hongze Cheng 已提交
1064 1065 1066
  SUdfAggRes     *udfRes = (SUdfAggRes *)GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
  udfRes->finalResBuf = (char *)udfRes + sizeof(SUdfAggRes);
  udfRes->interResBuf = (char *)udfRes + sizeof(SUdfAggRes) + session->outputLen;
S
shenglian zhou 已提交
1067

H
Hongze Cheng 已提交
1068 1069 1070 1071
  SInputColumnInfoData *pInput = &pCtx->input;
  int32_t               numOfCols = pInput->numOfInputCols;
  int32_t               start = pInput->startRowIndex;
  int32_t               numOfRows = pInput->numOfRows;
S
shenglian zhou 已提交
1072

H
Hongze Cheng 已提交
1073
  SSDataBlock *pTempBlock = createDataBlock();
1074 1075
  pTempBlock->info.rows = pInput->totalRows;
  pTempBlock->info.uid = pInput->uid;
S
shenglian zhou 已提交
1076
  for (int32_t i = 0; i < numOfCols; ++i) {
1077
    blockDataAppendColInfo(pTempBlock, pInput->pData[i]);
S
shenglian zhou 已提交
1078 1079
  }

1080
  SSDataBlock *inputBlock = blockDataExtractBlock(pTempBlock, start, numOfRows);
S
shenglian zhou 已提交
1081

H
Hongze Cheng 已提交
1082
  SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = session->bufSize, .numOfResult = udfRes->interResNum};
S
shenglian zhou 已提交
1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102
  SUdfInterBuf newState = {0};

  udfCode = doCallUdfAggProcess(session, inputBlock, &state, &newState);
  if (udfCode != 0) {
    fnError("udfAggProcess error. code: %d", udfCode);
    newState.numOfResult = 0;
  } else {
    udfRes->interResNum = newState.numOfResult;
    if (newState.bufLen <= session->bufSize) {
      memcpy(udfRes->interResBuf, newState.buf, newState.bufLen);
    } else {
      fnError("udfc inter buf size %d is greater than function bufSize %d", newState.bufLen, session->bufSize);
      udfCode = TSDB_CODE_UDF_INVALID_BUFSIZE;
    }
  }
  if (newState.numOfResult == 1 || state.numOfResult == 1) {
    GET_RES_INFO(pCtx)->numOfRes = 1;
  }

  blockDataDestroy(inputBlock);
1103 1104 1105

  taosArrayDestroy(pTempBlock->pDataBlock);
  taosMemoryFree(pTempBlock);
S
shenglian zhou 已提交
1106 1107 1108 1109 1110 1111

  releaseUdfFuncHandle(pCtx->udfName);
  freeUdfInterBuf(&newState);
  return udfCode;
}

H
Hongze Cheng 已提交
1112 1113
int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock) {
  int32_t        udfCode = 0;
S
shenglian zhou 已提交
1114 1115 1116 1117 1118 1119 1120
  UdfcFuncHandle handle = 0;
  if ((udfCode = acquireUdfFuncHandle((char *)pCtx->udfName, &handle)) != 0) {
    fnError("udfAggProcess  error. step acquireUdfFuncHandle. udf code: %d", udfCode);
    return udfCode;
  }

  SUdfcUvSession *session = handle;
H
Hongze Cheng 已提交
1121 1122 1123
  SUdfAggRes     *udfRes = (SUdfAggRes *)GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
  udfRes->finalResBuf = (char *)udfRes + sizeof(SUdfAggRes);
  udfRes->interResBuf = (char *)udfRes + sizeof(SUdfAggRes) + session->outputLen;
S
shenglian zhou 已提交
1124 1125

  SUdfInterBuf resultBuf = {0};
H
Hongze Cheng 已提交
1126 1127 1128
  SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = session->bufSize, .numOfResult = udfRes->interResNum};
  int32_t      udfCallCode = 0;
  udfCallCode = doCallUdfAggFinalize(session, &state, &resultBuf);
S
shenglian zhou 已提交
1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153
  if (udfCallCode != 0) {
    fnError("udfAggFinalize error. doCallUdfAggFinalize step. udf code:%d", udfCallCode);
    GET_RES_INFO(pCtx)->numOfRes = 0;
  } else {
    if (resultBuf.bufLen <= session->outputLen) {
      memcpy(udfRes->finalResBuf, resultBuf.buf, session->outputLen);
      udfRes->finalResNum = resultBuf.numOfResult;
      GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum;
    } else {
      fnError("udfc inter buf size %d is greater than function output size %d", resultBuf.bufLen, session->outputLen);
      GET_RES_INFO(pCtx)->numOfRes = 0;
      udfCallCode = TSDB_CODE_UDF_INVALID_OUTPUT_TYPE;
    }
  }

  freeUdfInterBuf(&resultBuf);

  int32_t numOfResults = functionFinalizeWithResultBuf(pCtx, pBlock, udfRes->finalResBuf);
  releaseUdfFuncHandle(pCtx->udfName);
  return udfCallCode == 0 ? numOfResults : udfCallCode;
}

void onUdfcPipeClose(uv_handle_t *handle) {
  SClientUvConn *conn = handle->data;
  if (!QUEUE_EMPTY(&conn->taskQueue)) {
H
Hongze Cheng 已提交
1154
    QUEUE             *h = QUEUE_HEAD(&conn->taskQueue);
S
shenglian zhou 已提交
1155 1156 1157 1158 1159
    SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
    task->errCode = 0;
    QUEUE_REMOVE(&task->procTaskQueue);
    uv_sem_post(&task->taskSem);
  }
1160 1161 1162
  if (conn->session != NULL) {
    conn->session->udfUvPipe = NULL;
  }
S
shenglian zhou 已提交
1163 1164
  taosMemoryFree(conn->readBuf.buf);
  taosMemoryFree(conn);
H
Hongze Cheng 已提交
1165
  taosMemoryFree((uv_pipe_t *)handle);
S
shenglian zhou 已提交
1166 1167 1168 1169 1170 1171 1172
}

int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask) {
  fnDebug("udfc get uv task result. task: %p, uvTask: %p", task, uvTask);
  if (uvTask->type == UV_TASK_REQ_RSP) {
    if (uvTask->rspBuf.base != NULL) {
      SUdfResponse rsp = {0};
H
Hongze Cheng 已提交
1173
      void        *buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp);
S
shenglian zhou 已提交
1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208
      assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base));
      task->errCode = rsp.code;

      switch (task->type) {
        case UDF_TASK_SETUP: {
          task->_setup.rsp = rsp.setupRsp;
          break;
        }
        case UDF_TASK_CALL: {
          task->_call.rsp = rsp.callRsp;
          break;
        }
        case UDF_TASK_TEARDOWN: {
          task->_teardown.rsp = rsp.teardownRsp;
          break;
        }
        default: {
          break;
        }
      }

      // TODO: the call buffer is setup and freed by udf invocation
      taosMemoryFree(uvTask->rspBuf.base);
    } else {
      task->errCode = uvTask->errCode;
    }
  } else if (uvTask->type == UV_TASK_CONNECT) {
    task->errCode = uvTask->errCode;
  } else if (uvTask->type == UV_TASK_DISCONNECT) {
    task->errCode = uvTask->errCode;
  }
  return 0;
}

void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
1209
  SClientUvConn  *conn = handle->data;
S
shenglian zhou 已提交
1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226
  SClientConnBuf *connBuf = &conn->readBuf;

  int32_t msgHeadSize = sizeof(int32_t) + sizeof(int64_t);
  if (connBuf->cap == 0) {
    connBuf->buf = taosMemoryMalloc(msgHeadSize);
    if (connBuf->buf) {
      connBuf->len = 0;
      connBuf->cap = msgHeadSize;
      connBuf->total = -1;

      buf->base = connBuf->buf;
      buf->len = connBuf->cap;
    } else {
      fnError("udfc allocate buffer failure. size: %d", msgHeadSize);
      buf->base = NULL;
      buf->len = 0;
    }
1227 1228 1229
  } else if (connBuf->total == -1 && connBuf->len < msgHeadSize) {
    buf->base = connBuf->buf + connBuf->len;
    buf->len = msgHeadSize - connBuf->len;
S
shenglian zhou 已提交
1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243
  } else {
    connBuf->cap = connBuf->total > connBuf->cap ? connBuf->total : connBuf->cap;
    void *resultBuf = taosMemoryRealloc(connBuf->buf, connBuf->cap);
    if (resultBuf) {
      connBuf->buf = resultBuf;
      buf->base = connBuf->buf + connBuf->len;
      buf->len = connBuf->cap - connBuf->len;
    } else {
      fnError("udfc re-allocate buffer failure. size: %d", connBuf->cap);
      buf->base = NULL;
      buf->len = 0;
    }
  }

1244
  fnDebug("udfc uv alloc buffer: cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total);
S
shenglian zhou 已提交
1245 1246 1247 1248
}

bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) {
  if (connBuf->total == -1 && connBuf->len >= sizeof(int32_t)) {
H
Hongze Cheng 已提交
1249
    connBuf->total = *(int32_t *)(connBuf->buf);
S
shenglian zhou 已提交
1250 1251
  }
  if (connBuf->len == connBuf->cap && connBuf->total == connBuf->cap) {
1252
    fnDebug("udfc complete message is received, now handle it");
S
shenglian zhou 已提交
1253 1254 1255 1256 1257 1258 1259
    return true;
  }
  return false;
}

void udfcUvHandleRsp(SClientUvConn *conn) {
  SClientConnBuf *connBuf = &conn->readBuf;
H
Hongze Cheng 已提交
1260
  int64_t         seqNum = *(int64_t *)(connBuf->buf + sizeof(int32_t));  // msglen then seqnum
S
shenglian zhou 已提交
1261 1262

  if (QUEUE_EMPTY(&conn->taskQueue)) {
H
Hongze Cheng 已提交
1263
    fnError("udfc no task waiting on connection. response seqnum:%" PRId64, seqNum);
S
shenglian zhou 已提交
1264 1265
    return;
  }
H
Hongze Cheng 已提交
1266
  bool               found = false;
S
shenglian zhou 已提交
1267
  SClientUvTaskNode *taskFound = NULL;
H
Hongze Cheng 已提交
1268
  QUEUE             *h = QUEUE_NEXT(&conn->taskQueue);
S
shenglian zhou 已提交
1269 1270 1271
  SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);

  while (h != &conn->taskQueue) {
1272
    fnDebug("udfc handle response iterate through queue. uvTask:%d-%p", task->seqNum, task);
S
shenglian zhou 已提交
1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300
    if (task->seqNum == seqNum) {
      if (found == false) {
        found = true;
        taskFound = task;
      } else {
        fnError("udfc more than one task waiting for the same response");
        continue;
      }
    }
    h = QUEUE_NEXT(h);
    task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
  }

  if (taskFound) {
    taskFound->rspBuf = uv_buf_init(connBuf->buf, connBuf->len);
    QUEUE_REMOVE(&taskFound->connTaskQueue);
    QUEUE_REMOVE(&taskFound->procTaskQueue);
    uv_sem_post(&taskFound->taskSem);
  } else {
    fnError("no task is waiting for the response.");
  }
  connBuf->buf = NULL;
  connBuf->total = -1;
  connBuf->len = 0;
  connBuf->cap = 0;
}

void udfcUvHandleError(SClientUvConn *conn) {
1301
  fnDebug("handle error on conn: %p, pipe: %p", conn, conn->pipe);
S
shenglian zhou 已提交
1302
  while (!QUEUE_EMPTY(&conn->taskQueue)) {
H
Hongze Cheng 已提交
1303
    QUEUE             *h = QUEUE_HEAD(&conn->taskQueue);
S
shenglian zhou 已提交
1304 1305 1306 1307 1308 1309 1310
    SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
    task->errCode = TSDB_CODE_UDF_PIPE_READ_ERR;
    QUEUE_REMOVE(&task->connTaskQueue);
    QUEUE_REMOVE(&task->procTaskQueue);
    uv_sem_post(&task->taskSem);
  }

H
Hongze Cheng 已提交
1311
  uv_close((uv_handle_t *)conn->pipe, onUdfcPipeClose);
S
shenglian zhou 已提交
1312 1313 1314
}

void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
1315
  fnDebug("udfc client %p, client read from pipe. nread: %zd", client, nread);
S
shenglian zhou 已提交
1316 1317
  if (nread == 0) return;

H
Hongze Cheng 已提交
1318
  SClientUvConn  *conn = client->data;
S
shenglian zhou 已提交
1319 1320 1321 1322 1323 1324 1325 1326
  SClientConnBuf *connBuf = &conn->readBuf;
  if (nread > 0) {
    connBuf->len += nread;
    if (isUdfcUvMsgComplete(connBuf)) {
      udfcUvHandleRsp(conn);
    }
  }
  if (nread < 0) {
1327
    fnError("udfc client pipe %p read error: %zd(%s).", client, nread, uv_strerror(nread));
S
shenglian zhou 已提交
1328 1329 1330 1331
    if (nread == UV_EOF) {
      fnError("\tudfc client pipe %p closed", client);
    }
    udfcUvHandleError(conn);
1332 1333 1334
  }
}

1335 1336 1337 1338
void onUdfcPipeWrite(uv_write_t *write, int status) {
  SClientUvConn *conn = write->data;
  if (status < 0) {
    fnError("udfc client connection %p write failed. status: %d(%s)", conn, status, uv_strerror(status));
1339
    udfcUvHandleError(conn);
1340 1341
  } else {
    fnDebug("udfc client connection %p write succeed", conn);
1342
  }
wafwerar's avatar
wafwerar 已提交
1343
  taosMemoryFree(write);
1344
}
H
Haojun Liao 已提交
1345

1346
void onUdfcPipeConnect(uv_connect_t *connect, int status) {
1347 1348
  SClientUvTaskNode *uvTask = connect->data;
  if (status != 0) {
H
Hongze Cheng 已提交
1349
    fnError("client connect error, task seq: %" PRId64 ", code: %s", uvTask->seqNum, uv_strerror(status));
H
Haojun Liao 已提交
1350
  }
1351 1352 1353
  uvTask->errCode = status;

  uv_read_start((uv_stream_t *)uvTask->pipe, udfcAllocateBuffer, onUdfcPipeRead);
wafwerar's avatar
wafwerar 已提交
1354
  taosMemoryFree(connect);
S
shenglian zhou 已提交
1355
  QUEUE_REMOVE(&uvTask->procTaskQueue);
1356
  uv_sem_post(&uvTask->taskSem);
1357
}
H
Haojun Liao 已提交
1358

S
shenglian zhou 已提交
1359
int32_t udfcInitializeUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskNode *uvTask) {
1360
  uvTask->type = uvTaskType;
1361
  uvTask->udfc = task->session->udfc;
1362 1363 1364

  if (uvTaskType == UV_TASK_CONNECT) {
  } else if (uvTaskType == UV_TASK_REQ_RSP) {
S
slzhou 已提交
1365
    uvTask->pipe = task->session->udfUvPipe;
1366 1367
    SUdfRequest request;
    request.type = task->type;
H
Hongze Cheng 已提交
1368
    request.seqNum = atomic_fetch_add_64(&gUdfTaskSeqNum, 1);
1369 1370

    if (task->type == UDF_TASK_SETUP) {
S
shenglian zhou 已提交
1371
      request.setup = task->_setup.req;
1372 1373
      request.type = UDF_TASK_SETUP;
    } else if (task->type == UDF_TASK_CALL) {
S
shenglian zhou 已提交
1374
      request.call = task->_call.req;
1375 1376
      request.type = UDF_TASK_CALL;
    } else if (task->type == UDF_TASK_TEARDOWN) {
S
shenglian zhou 已提交
1377
      request.teardown = task->_teardown.req;
1378 1379
      request.type = UDF_TASK_TEARDOWN;
    } else {
S
shenglian zhou 已提交
1380
      fnError("udfc create uv task, invalid task type : %d", task->type);
1381
    }
1382 1383
    int32_t bufLen = encodeUdfRequest(NULL, &request);
    request.msgLen = bufLen;
S
slzhou 已提交
1384 1385
    void *bufBegin = taosMemoryMalloc(bufLen);
    void *buf = bufBegin;
1386
    encodeUdfRequest(&buf, &request);
S
slzhou 已提交
1387
    uvTask->reqBuf = uv_buf_init(bufBegin, bufLen);
1388 1389
    uvTask->seqNum = request.seqNum;
  } else if (uvTaskType == UV_TASK_DISCONNECT) {
S
slzhou 已提交
1390
    uvTask->pipe = task->session->udfUvPipe;
1391 1392
  }
  uv_sem_init(&uvTask->taskSem, 0);
H
Haojun Liao 已提交
1393

1394 1395
  return 0;
}
H
Haojun Liao 已提交
1396

S
slzhou 已提交
1397
int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask) {
1398
  fnDebug("queue uv task to event loop, uvTask: %d-%p", uvTask->type, uvTask);
1399 1400 1401 1402 1403
  SUdfcProxy *udfc = uvTask->udfc;
  uv_mutex_lock(&udfc->taskQueueMutex);
  QUEUE_INSERT_TAIL(&udfc->taskQueue, &uvTask->recvTaskQueue);
  uv_mutex_unlock(&udfc->taskQueueMutex);
  uv_async_send(&udfc->loopTaskAync);
H
Haojun Liao 已提交
1404

1405
  uv_sem_wait(&uvTask->taskSem);
H
Hongze Cheng 已提交
1406
  fnInfo("udfc uvTask finished. uvTask:%" PRId64 "-%d-%p", uvTask->seqNum, uvTask->type, uvTask);
1407
  uv_sem_destroy(&uvTask->taskSem);
H
Haojun Liao 已提交
1408

1409 1410
  return 0;
}
H
Haojun Liao 已提交
1411

S
slzhou 已提交
1412
int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
H
Hongze Cheng 已提交
1413
  fnDebug("event loop start uv task. uvTask: %" PRId64 "-%d-%p", uvTask->seqNum, uvTask->type, uvTask);
1414 1415
  int32_t code = 0;

1416 1417
  switch (uvTask->type) {
    case UV_TASK_CONNECT: {
wafwerar's avatar
wafwerar 已提交
1418
      uv_pipe_t *pipe = taosMemoryMalloc(sizeof(uv_pipe_t));
1419
      uv_pipe_init(&uvTask->udfc->uvLoop, pipe, 0);
1420
      uvTask->pipe = pipe;
H
Haojun Liao 已提交
1421

S
slzhou 已提交
1422
      SClientUvConn *conn = taosMemoryCalloc(1, sizeof(SClientUvConn));
1423 1424 1425 1426 1427
      conn->pipe = pipe;
      conn->readBuf.len = 0;
      conn->readBuf.cap = 0;
      conn->readBuf.buf = 0;
      conn->readBuf.total = -1;
S
shenglian zhou 已提交
1428
      QUEUE_INIT(&conn->taskQueue);
H
Haojun Liao 已提交
1429

1430 1431
      pipe->data = conn;

wafwerar's avatar
wafwerar 已提交
1432
      uv_connect_t *connReq = taosMemoryMalloc(sizeof(uv_connect_t));
1433
      connReq->data = uvTask;
1434
      uv_pipe_connect(connReq, pipe, uvTask->udfc->udfdPipeName, onUdfcPipeConnect);
1435
      code = 0;
H
Haojun Liao 已提交
1436
      break;
1437 1438 1439
    }
    case UV_TASK_REQ_RSP: {
      uv_pipe_t *pipe = uvTask->pipe;
1440 1441 1442 1443
      if (pipe == NULL) {
        code = TSDB_CODE_UDF_PIPE_NO_PIPE;
      } else {
        uv_write_t *write = taosMemoryMalloc(sizeof(uv_write_t));
1444
        write->data = pipe->data;
H
Hongze Cheng 已提交
1445
        QUEUE *connTaskQueue = &((SClientUvConn *)pipe->data)->taskQueue;
1446 1447
        QUEUE_INSERT_TAIL(connTaskQueue, &uvTask->connTaskQueue);
        int err = uv_write(write, (uv_stream_t *)pipe, &uvTask->reqBuf, 1, onUdfcPipeWrite);
1448
        if (err != 0) {
S
slzhou 已提交
1449
          taosMemoryFree(write);
1450
          fnError("udfc event loop start req_rsp task uv_write failed. uvtask: %p, code: %s", uvTask, uv_strerror(err));
1451 1452
        }
        code = err;
1453
      }
1454 1455 1456
      break;
    }
    case UV_TASK_DISCONNECT: {
1457 1458 1459 1460 1461 1462 1463 1464 1465
      uv_pipe_t *pipe = uvTask->pipe;
      if (pipe == NULL) {
        code = TSDB_CODE_UDF_PIPE_NO_PIPE;
      } else {
        SClientUvConn *conn = pipe->data;
        QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
        uv_close((uv_handle_t *)uvTask->pipe, onUdfcPipeClose);
        code = 0;
      }
1466 1467 1468
      break;
    }
    default: {
H
Hongze Cheng 已提交
1469
      fnError("udfc event loop unknown task type.") break;
1470 1471
    }
  }
H
Haojun Liao 已提交
1472

1473
  return code;
1474
}
H
Haojun Liao 已提交
1475

1476
void udfcAsyncTaskCb(uv_async_t *async) {
1477
  SUdfcProxy *udfc = async->data;
H
Hongze Cheng 已提交
1478
  QUEUE       wq;
1479

1480 1481 1482
  uv_mutex_lock(&udfc->taskQueueMutex);
  QUEUE_MOVE(&udfc->taskQueue, &wq);
  uv_mutex_unlock(&udfc->taskQueueMutex);
1483

S
shenglian zhou 已提交
1484
  while (!QUEUE_EMPTY(&wq)) {
H
Hongze Cheng 已提交
1485
    QUEUE *h = QUEUE_HEAD(&wq);
S
shenglian zhou 已提交
1486 1487
    QUEUE_REMOVE(h);
    SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, recvTaskQueue);
H
Hongze Cheng 已提交
1488
    int32_t            code = udfcStartUvTask(task);
1489 1490
    if (code == 0) {
      QUEUE_INSERT_TAIL(&udfc->uvProcTaskQueue, &task->procTaskQueue);
1491 1492 1493
    } else {
      task->errCode = code;
      uv_sem_post(&task->taskSem);
1494
    }
1495 1496 1497
  }
}

1498
void cleanUpUvTasks(SUdfcProxy *udfc) {
H
Hongze Cheng 已提交
1499
  fnDebug("clean up uv tasks") QUEUE wq;
1500

1501 1502 1503
  uv_mutex_lock(&udfc->taskQueueMutex);
  QUEUE_MOVE(&udfc->taskQueue, &wq);
  uv_mutex_unlock(&udfc->taskQueueMutex);
1504

S
shenglian zhou 已提交
1505
  while (!QUEUE_EMPTY(&wq)) {
H
Hongze Cheng 已提交
1506
    QUEUE *h = QUEUE_HEAD(&wq);
S
shenglian zhou 已提交
1507 1508
    QUEUE_REMOVE(h);
    SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, recvTaskQueue);
1509
    if (udfc->udfcState == UDFC_STATE_STOPPING) {
1510
      task->errCode = TSDB_CODE_UDF_STOPPING;
1511 1512 1513 1514
    }
    uv_sem_post(&task->taskSem);
  }

1515
  while (!QUEUE_EMPTY(&udfc->uvProcTaskQueue)) {
H
Hongze Cheng 已提交
1516
    QUEUE *h = QUEUE_HEAD(&udfc->uvProcTaskQueue);
S
shenglian zhou 已提交
1517 1518
    QUEUE_REMOVE(h);
    SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, procTaskQueue);
1519
    if (udfc->udfcState == UDFC_STATE_STOPPING) {
1520
      task->errCode = TSDB_CODE_UDF_STOPPING;
S
shenglian zhou 已提交
1521 1522 1523 1524
    }
    uv_sem_post(&task->taskSem);
  }
}
1525

S
shenglian zhou 已提交
1526
void udfStopAsyncCb(uv_async_t *async) {
1527
  SUdfcProxy *udfc = async->data;
1528
  cleanUpUvTasks(udfc);
1529 1530
  if (udfc->udfcState == UDFC_STATE_STOPPING) {
    uv_stop(&udfc->uvLoop);
S
shenglian zhou 已提交
1531
  }
1532
}
S
shenglian zhou 已提交
1533

S
shenglian zhou 已提交
1534
void constructUdfService(void *argsThread) {
1535 1536 1537
  SUdfcProxy *udfc = (SUdfcProxy *)argsThread;
  uv_loop_init(&udfc->uvLoop);

1538
  uv_async_init(&udfc->uvLoop, &udfc->loopTaskAync, udfcAsyncTaskCb);
1539 1540 1541 1542 1543 1544 1545
  udfc->loopTaskAync.data = udfc;
  uv_async_init(&udfc->uvLoop, &udfc->loopStopAsync, udfStopAsyncCb);
  udfc->loopStopAsync.data = udfc;
  uv_mutex_init(&udfc->taskQueueMutex);
  QUEUE_INIT(&udfc->taskQueue);
  QUEUE_INIT(&udfc->uvProcTaskQueue);
  uv_barrier_wait(&udfc->initBarrier);
H
Hongze Cheng 已提交
1546
  // TODO return value of uv_run
1547 1548
  uv_run(&udfc->uvLoop, UV_RUN_DEFAULT);
  uv_loop_close(&udfc->uvLoop);
S
slzhou 已提交
1549 1550 1551 1552

  uv_walk(&udfc->uvLoop, udfUdfdCloseWalkCb, NULL);
  uv_run(&udfc->uvLoop, UV_RUN_DEFAULT);
  uv_loop_close(&udfc->uvLoop);
1553 1554
}

1555 1556 1557 1558 1559
int32_t udfcOpen() {
  int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 0, 1);
  if (old == 1) {
    return 0;
  }
1560
  SUdfcProxy *proxy = &gUdfdProxy;
1561
  getUdfdPipeName(proxy->udfdPipeName, sizeof(proxy->udfdPipeName));
1562 1563 1564 1565 1566 1567
  proxy->udfcState = UDFC_STATE_STARTNG;
  uv_barrier_init(&proxy->initBarrier, 2);
  uv_thread_create(&proxy->loopThread, constructUdfService, proxy);
  atomic_store_8(&proxy->udfcState, UDFC_STATE_READY);
  proxy->udfcState = UDFC_STATE_READY;
  uv_barrier_wait(&proxy->initBarrier);
1568 1569
  uv_mutex_init(&proxy->udfStubsMutex);
  proxy->udfStubs = taosArrayInit(8, sizeof(SUdfcFuncStub));
H
Hongze Cheng 已提交
1570
  fnInfo("udfc initialized") return 0;
1571 1572
}

1573 1574 1575 1576 1577 1578
int32_t udfcClose() {
  int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 1, 0);
  if (old == 0) {
    return 0;
  }

1579 1580 1581 1582 1583 1584
  SUdfcProxy *udfc = &gUdfdProxy;
  udfc->udfcState = UDFC_STATE_STOPPING;
  uv_async_send(&udfc->loopStopAsync);
  uv_thread_join(&udfc->loopThread);
  uv_mutex_destroy(&udfc->taskQueueMutex);
  uv_barrier_destroy(&udfc->initBarrier);
1585 1586
  taosArrayDestroy(udfc->udfStubs);
  uv_mutex_destroy(&udfc->udfStubsMutex);
1587
  udfc->udfcState = UDFC_STATE_INITAL;
S
Shengliang Guan 已提交
1588
  fnInfo("udfc is cleaned up");
1589 1590 1591
  return 0;
}

S
slzhou 已提交
1592
int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) {
S
shenglian zhou 已提交
1593
  SClientUvTaskNode *uvTask = taosMemoryCalloc(1, sizeof(SClientUvTaskNode));
1594
  fnDebug("udfc client task: %p created uvTask: %p. pipe: %p", task, uvTask, task->session->udfUvPipe);
S
shenglian zhou 已提交
1595 1596

  udfcInitializeUvTask(task, uvTaskType, uvTask);
S
slzhou 已提交
1597 1598
  udfcQueueUvTask(uvTask);
  udfcGetUdfTaskResultFromUvTask(task, uvTask);
1599
  if (uvTaskType == UV_TASK_CONNECT) {
S
slzhou 已提交
1600 1601 1602
    task->session->udfUvPipe = uvTask->pipe;
    SClientUvConn *conn = uvTask->pipe->data;
    conn->session = task->session;
S
slzhou 已提交
1603
  }
1604 1605
  taosMemoryFree(uvTask->reqBuf.base);
  uvTask->reqBuf.base = NULL;
S
shenglian zhou 已提交
1606
  taosMemoryFree(uvTask);
S
shenglian zhou 已提交
1607 1608
  fnDebug("udfc freed uvTask: %p", task);

S
shenglian zhou 已提交
1609 1610
  uvTask = NULL;
  return task->errCode;
S
slzhou 已提交
1611 1612
}

S
shenglian zhou 已提交
1613 1614 1615
int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
  if (gUdfdProxy.udfcState != UDFC_STATE_READY) {
    return TSDB_CODE_UDF_INVALID_STATE;
1616
  }
H
Hongze Cheng 已提交
1617
  SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
S
shenglian zhou 已提交
1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628
  task->errCode = 0;
  task->session = taosMemoryCalloc(1, sizeof(SUdfcUvSession));
  task->session->udfc = &gUdfdProxy;
  task->type = UDF_TASK_SETUP;

  SUdfSetupRequest *req = &task->_setup.req;
  strncpy(req->udfName, udfName, TSDB_FUNC_NAME_LEN);

  int32_t errCode = udfcRunUdfUvTask(task, UV_TASK_CONNECT);
  if (errCode != 0) {
    fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName);
1629 1630
    taosMemoryFree(task->session);
    taosMemoryFree(task);
S
shenglian zhou 已提交
1631
    return TSDB_CODE_UDF_PIPE_CONNECT_ERR;
1632
  }
S
slzhou 已提交
1633

S
shenglian zhou 已提交
1634 1635 1636 1637 1638 1639 1640
  udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);

  SUdfSetupResponse *rsp = &task->_setup.rsp;
  task->session->severHandle = rsp->udfHandle;
  task->session->outputType = rsp->outputType;
  task->session->outputLen = rsp->outputLen;
  task->session->bufSize = rsp->bufSize;
S
slzhou 已提交
1641
  strncpy(task->session->udfName, udfName, TSDB_FUNC_NAME_LEN);
S
shenglian zhou 已提交
1642 1643 1644 1645 1646
  if (task->errCode != 0) {
    fnError("failed to setup udf. udfname: %s, err: %d", udfName, task->errCode)
  } else {
    fnInfo("sucessfully setup udf func handle. udfName: %s, handle: %p", udfName, task->session);
    *funcHandle = task->session;
S
slzhou 已提交
1647
  }
S
shenglian zhou 已提交
1648 1649 1650
  int32_t err = task->errCode;
  taosMemoryFree(task);
  return err;
S
slzhou 已提交
1651 1652
}

1653
int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdfInterBuf *state, SUdfInterBuf *state2,
H
Hongze Cheng 已提交
1654
                SSDataBlock *output, SUdfInterBuf *newState) {
1655
  fnDebug("udfc call udf. callType: %d, funcHandle: %p", callType, handle);
H
Hongze Cheng 已提交
1656
  SUdfcUvSession *session = (SUdfcUvSession *)handle;
S
slzhou 已提交
1657 1658
  if (session->udfUvPipe == NULL) {
    fnError("No pipe to udfd");
1659
    return TSDB_CODE_UDF_PIPE_NO_PIPE;
S
slzhou 已提交
1660 1661
  }
  SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
1662
  task->errCode = 0;
H
Hongze Cheng 已提交
1663
  task->session = (SUdfcUvSession *)handle;
1664 1665 1666
  task->type = UDF_TASK_CALL;

  SUdfCallRequest *req = &task->_call.req;
S
slzhou 已提交
1667
  req->udfHandle = task->session->severHandle;
S
slzhou 已提交
1668
  req->callType = callType;
S
slzhou 已提交
1669

S
shenglian zhou 已提交
1670
  switch (callType) {
1671 1672 1673 1674
    case TSDB_UDF_CALL_AGG_INIT: {
      req->initFirst = 1;
      break;
    }
S
shenglian zhou 已提交
1675 1676 1677 1678 1679
    case TSDB_UDF_CALL_AGG_PROC: {
      req->block = *input;
      req->interBuf = *state;
      break;
    }
1680 1681 1682 1683 1684 1685
    case TSDB_UDF_CALL_AGG_MERGE: {
      req->interBuf = *state;
      req->interBuf2 = *state2;
      break;
    }
    case TSDB_UDF_CALL_AGG_FIN: {
S
shenglian zhou 已提交
1686 1687 1688 1689 1690 1691 1692 1693 1694
      req->interBuf = *state;
      break;
    }
    case TSDB_UDF_CALL_SCALA_PROC: {
      req->block = *input;
      break;
    }
  }

S
slzhou 已提交
1695
  udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);
1696

1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721
  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;
      }
S
shenglian zhou 已提交
1722
    }
S
slzhou 已提交
1723 1724
  };
  int err = task->errCode;
wafwerar's avatar
wafwerar 已提交
1725
  taosMemoryFree(task);
S
slzhou 已提交
1726
  return err;
1727 1728
}

1729
int32_t doCallUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf) {
S
slzhou 已提交
1730 1731 1732 1733 1734 1735 1736 1737 1738
  int8_t callType = TSDB_UDF_CALL_AGG_INIT;

  int32_t err = callUdf(handle, callType, NULL, NULL, NULL, NULL, interBuf);

  return err;
}

// input: block, state
// output: interbuf,
1739
int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInterBuf *state, SUdfInterBuf *newState) {
H
Hongze Cheng 已提交
1740
  int8_t  callType = TSDB_UDF_CALL_AGG_PROC;
S
slzhou 已提交
1741 1742 1743 1744 1745 1746
  int32_t err = callUdf(handle, callType, block, state, NULL, NULL, newState);
  return err;
}

// input: interbuf1, interbuf2
// output: resultBuf
H
Hongze Cheng 已提交
1747 1748 1749
int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2,
                          SUdfInterBuf *resultBuf) {
  int8_t  callType = TSDB_UDF_CALL_AGG_MERGE;
S
slzhou 已提交
1750 1751 1752 1753 1754 1755
  int32_t err = callUdf(handle, callType, NULL, interBuf1, interBuf2, NULL, resultBuf);
  return err;
}

// input: interBuf
// output: resultData
1756
int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData) {
H
Hongze Cheng 已提交
1757
  int8_t  callType = TSDB_UDF_CALL_AGG_FIN;
S
slzhou 已提交
1758 1759 1760 1761
  int32_t err = callUdf(handle, callType, NULL, interBuf, NULL, NULL, resultData);
  return err;
}

H
Hongze Cheng 已提交
1762 1763
int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output) {
  int8_t      callType = TSDB_UDF_CALL_SCALA_PROC;
S
slzhou 已提交
1764 1765 1766
  SSDataBlock inputBlock = {0};
  convertScalarParamToDataBlock(input, numOfCols, &inputBlock);
  SSDataBlock resultBlock = {0};
H
Hongze Cheng 已提交
1767
  int32_t     err = callUdf(handle, callType, &inputBlock, NULL, NULL, &resultBlock, NULL);
S
slzhou 已提交
1768 1769
  if (err == 0) {
    convertDataBlockToScalarParm(&resultBlock, output);
S
slzhou 已提交
1770
    taosArrayDestroy(resultBlock.pDataBlock);
S
slzhou 已提交
1771
  }
S
slzhou 已提交
1772 1773

  taosArrayDestroy(inputBlock.pDataBlock);
S
slzhou 已提交
1774 1775 1776
  return err;
}

1777
int32_t doTeardownUdf(UdfcFuncHandle handle) {
H
Hongze Cheng 已提交
1778
  SUdfcUvSession *session = (SUdfcUvSession *)handle;
S
slzhou 已提交
1779

S
slzhou 已提交
1780
  if (session->udfUvPipe == NULL) {
S
slzhou 已提交
1781
    fnError("tear down udf. pipe to udfd does not exist. udf name: %s", session->udfName);
1782
    taosMemoryFree(session);
1783
    return TSDB_CODE_UDF_PIPE_NO_PIPE;
S
slzhou 已提交
1784 1785 1786
  }

  SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
1787
  task->errCode = 0;
S
slzhou 已提交
1788
  task->session = session;
1789 1790 1791 1792 1793
  task->type = UDF_TASK_TEARDOWN;

  SUdfTeardownRequest *req = &task->_teardown.req;
  req->udfHandle = task->session->severHandle;

S
slzhou 已提交
1794
  udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);
1795 1796 1797

  int32_t err = task->errCode;

S
slzhou 已提交
1798
  udfcRunUdfUvTask(task, UV_TASK_DISCONNECT);
1799

S
slzhou 已提交
1800
  fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
H
Hongze Cheng 已提交
1801
  // TODO: synchronization refactor between libuv event loop and request thread
S
slzhou 已提交
1802
  if (session->udfUvPipe != NULL && session->udfUvPipe->data != NULL) {
1803 1804 1805
    SClientUvConn *conn = session->udfUvPipe->data;
    conn->session = NULL;
  }
1806
  taosMemoryFree(session);
wafwerar's avatar
wafwerar 已提交
1807
  taosMemoryFree(task);
1808 1809 1810

  return err;
}