transCli.c 16.3 KB
Newer Older
dengyihao's avatar
dengyihao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
/*
 * 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/>.
 */

#ifdef USE_UV

#include "transComm.h"

20
#define CONN_PERSIST_TIME(para) (para * 1000 * 10)
dengyihao's avatar
dengyihao 已提交
21

dengyihao's avatar
dengyihao 已提交
22 23 24
typedef struct SCliConn {
  uv_connect_t connReq;
  uv_stream_t* stream;
dengyihao's avatar
dengyihao 已提交
25
  uv_write_t*  writeReq;
dengyihao's avatar
dengyihao 已提交
26
  void*        hostThrd;
dengyihao's avatar
dengyihao 已提交
27
  SConnBuffer  readBuf;
dengyihao's avatar
dengyihao 已提交
28 29
  void*        data;
  queue        conn;
dengyihao's avatar
dengyihao 已提交
30 31
  char         spi;
  char         secured;
dengyihao's avatar
dengyihao 已提交
32
  uint64_t     expireTime;
dengyihao's avatar
dengyihao 已提交
33
  int8_t       notifyCount;  // timers already notify to client
dengyihao's avatar
dengyihao 已提交
34
  int32_t      ref;
dengyihao's avatar
dengyihao 已提交
35
} SCliConn;
dengyihao's avatar
dengyihao 已提交
36

dengyihao's avatar
dengyihao 已提交
37
typedef struct SCliMsg {
dengyihao's avatar
dengyihao 已提交
38 39 40 41
  STransConnCtx* ctx;
  SRpcMsg        msg;
  queue          q;
  uint64_t       st;
dengyihao's avatar
dengyihao 已提交
42 43 44 45 46 47
} SCliMsg;

typedef struct SCliThrdObj {
  pthread_t       thread;
  uv_loop_t*      loop;
  uv_async_t*     cliAsync;  //
dengyihao's avatar
dengyihao 已提交
48
  uv_timer_t*     pTimer;
dengyihao's avatar
dengyihao 已提交
49
  void*           pool;  // conn pool
dengyihao's avatar
dengyihao 已提交
50 51
  queue           msg;
  pthread_mutex_t msgMtx;
dengyihao's avatar
dengyihao 已提交
52
  uint64_t        nextTimeout;  // next timeout
dengyihao's avatar
dengyihao 已提交
53
  void*           pTransInst;   //
dengyihao's avatar
dengyihao 已提交
54

dengyihao's avatar
dengyihao 已提交
55 56 57 58 59 60 61 62 63
} SCliThrdObj;

typedef struct SClientObj {
  char          label[TSDB_LABEL_LEN];
  int32_t       index;
  int           numOfThreads;
  SCliThrdObj** pThreadObj;
} SClientObj;

dengyihao's avatar
dengyihao 已提交
64 65 66 67
typedef struct SConnList {
  queue conn;
} SConnList;

dengyihao's avatar
dengyihao 已提交
68
// conn pool
dengyihao's avatar
dengyihao 已提交
69
// add expire timeout and capacity limit
70 71
static void*     creatConnPool(int size);
static void*     destroyConnPool(void* pool);
dengyihao's avatar
dengyihao 已提交
72 73
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port);
static void      addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn);
dengyihao's avatar
dengyihao 已提交
74

dengyihao's avatar
dengyihao 已提交
75 76
// register timer in each thread to clear expire conn
static void clientTimeoutCb(uv_timer_t* handle);
dengyihao's avatar
dengyihao 已提交
77 78 79
// check whether already read complete packet from server
static bool clientReadComplete(SConnBuffer* pBuf);
// alloc buf for read
dengyihao's avatar
dengyihao 已提交
80
static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
81
// callback after read nbytes from socket
dengyihao's avatar
dengyihao 已提交
82
static void clientReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
83
// callback after write data to socket
dengyihao's avatar
dengyihao 已提交
84
static void clientWriteCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
85
// callback after conn  to server
dengyihao's avatar
dengyihao 已提交
86
static void clientConnCb(uv_connect_t* req, int status);
dengyihao's avatar
dengyihao 已提交
87
static void clientAsyncCb(uv_async_t* handle);
dengyihao's avatar
dengyihao 已提交
88
static void clientDestroy(uv_handle_t* handle);
89
static void clientConnDestroy(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
dengyihao's avatar
dengyihao 已提交
90

dengyihao's avatar
dengyihao 已提交
91 92 93 94
// process data read from server, auth/decompress etc later
static void clientHandleResp(SCliConn* conn);
// handle except about conn
static void clientHandleExcept(SCliConn* conn);
95 96
// handle req from app
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
dengyihao's avatar
dengyihao 已提交
97

dengyihao's avatar
dengyihao 已提交
98 99
static void clientMsgDestroy(SCliMsg* pMsg);
static void destroyTransConnCtx(STransConnCtx* ctx);
dengyihao's avatar
dengyihao 已提交
100 101 102 103
// thread obj
static SCliThrdObj* createThrdObj();
static void         destroyThrdObj(SCliThrdObj* pThrd);
// thread
dengyihao's avatar
dengyihao 已提交
104 105
static void* clientThread(void* arg);

106
static void clientHandleResp(SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
107
  STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx;
dengyihao's avatar
dengyihao 已提交
108
  SRpcInfo*      pRpc = pCtx->pTransInst;
dengyihao's avatar
dengyihao 已提交
109

dengyihao's avatar
dengyihao 已提交
110 111 112 113
  STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
  pHead->code = htonl(pHead->code);
  pHead->msgLen = htonl(pHead->msgLen);

dengyihao's avatar
dengyihao 已提交
114
  SRpcMsg rpcMsg;
dengyihao's avatar
dengyihao 已提交
115
  rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
dengyihao's avatar
dengyihao 已提交
116
  rpcMsg.pCont = transContFromHead((char*)pHead);
dengyihao's avatar
dengyihao 已提交
117 118
  rpcMsg.code = pHead->code;
  rpcMsg.msgType = pHead->msgType;
dengyihao's avatar
dengyihao 已提交
119
  rpcMsg.ahandle = pCtx->ahandle;
dengyihao's avatar
dengyihao 已提交
120

dengyihao's avatar
dengyihao 已提交
121
  tDebug("conn %p handle resp", conn);
dengyihao's avatar
dengyihao 已提交
122
  (pRpc->cfp)(NULL, &rpcMsg, NULL);
dengyihao's avatar
dengyihao 已提交
123
  conn->notifyCount += 1;
dengyihao's avatar
dengyihao 已提交
124 125

  SCliThrdObj* pThrd = conn->hostThrd;
dengyihao's avatar
dengyihao 已提交
126
  tfree(conn->data);
dengyihao's avatar
dengyihao 已提交
127 128 129
  // buf alread translated to rpcMsg.pCont
  transClearBuffer(&conn->readBuf);

dengyihao's avatar
dengyihao 已提交
130
  uv_read_start((uv_stream_t*)conn->stream, clientAllocBufferCb, clientReadCb);
dengyihao's avatar
dengyihao 已提交
131
  addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
dengyihao's avatar
dengyihao 已提交
132 133

  // start thread's timer of conn pool if not active
dengyihao's avatar
dengyihao 已提交
134 135 136
  if (!uv_is_active((uv_handle_t*)pThrd->pTimer) && pRpc->idleTime > 0) {
    uv_timer_start((uv_timer_t*)pThrd->pTimer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
  }
dengyihao's avatar
dengyihao 已提交
137 138 139
  destroyTransConnCtx(pCtx);
}
static void clientHandleExcept(SCliConn* pConn) {
dengyihao's avatar
dengyihao 已提交
140 141 142 143 144
  if (pConn->data == NULL) {
    clientConnDestroy(pConn, true);
    return;
  }
  tDebug("conn %p destroy", pConn);
dengyihao's avatar
dengyihao 已提交
145
  SCliMsg* pMsg = pConn->data;
dengyihao's avatar
dengyihao 已提交
146 147
  transFreeMsg((pMsg->msg.pCont));
  pMsg->msg.pCont = NULL;
dengyihao's avatar
dengyihao 已提交
148 149 150 151

  STransConnCtx* pCtx = pMsg->ctx;
  SRpcInfo*      pRpc = pCtx->pTransInst;

dengyihao's avatar
dengyihao 已提交
152
  SRpcMsg rpcMsg = {0};
dengyihao's avatar
dengyihao 已提交
153
  rpcMsg.ahandle = pCtx->ahandle;
dengyihao's avatar
dengyihao 已提交
154
  rpcMsg.code = -1;
dengyihao's avatar
dengyihao 已提交
155 156
  // SRpcInfo* pRpc = pMsg->ctx->pRpc;
  (pRpc->cfp)(NULL, &rpcMsg, NULL);
dengyihao's avatar
dengyihao 已提交
157
  tfree(pConn->data);
dengyihao's avatar
dengyihao 已提交
158 159 160
  pConn->notifyCount += 1;
  destroyTransConnCtx(pCtx);
  clientConnDestroy(pConn, true);
dengyihao's avatar
dengyihao 已提交
161
}
dengyihao's avatar
dengyihao 已提交
162

dengyihao's avatar
dengyihao 已提交
163 164
static void clientTimeoutCb(uv_timer_t* handle) {
  SCliThrdObj* pThrd = handle->data;
dengyihao's avatar
dengyihao 已提交
165
  SRpcInfo*    pRpc = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
166
  int64_t      currentTime = pThrd->nextTimeout;
dengyihao's avatar
dengyihao 已提交
167
  tDebug("timeout, try to remove expire conn from conn pool");
dengyihao's avatar
dengyihao 已提交
168

dengyihao's avatar
dengyihao 已提交
169
  SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
dengyihao's avatar
dengyihao 已提交
170 171 172 173 174 175
  while (p != NULL) {
    while (!QUEUE_IS_EMPTY(&p->conn)) {
      queue*    h = QUEUE_HEAD(&p->conn);
      SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
      if (c->expireTime < currentTime) {
        QUEUE_REMOVE(h);
176 177 178
        // uv_stream_t stm = *(c->stream);
        // uv_close((uv_handle_t*)&stm, clientDestroy);
        clientConnDestroy(c, true);
dengyihao's avatar
dengyihao 已提交
179 180 181 182
      } else {
        break;
      }
    }
dengyihao's avatar
dengyihao 已提交
183
    p = taosHashIterate((SHashObj*)pThrd->pool, p);
dengyihao's avatar
dengyihao 已提交
184 185
  }

dengyihao's avatar
dengyihao 已提交
186 187
  pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
  uv_timer_start(handle, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
dengyihao's avatar
dengyihao 已提交
188
}
189 190 191
static void* creatConnPool(int size) {
  // thread local, no lock
  return taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
dengyihao's avatar
dengyihao 已提交
192
}
193
static void* destroyConnPool(void* pool) {
dengyihao's avatar
dengyihao 已提交
194
  SConnList* connList = taosHashIterate((SHashObj*)pool, NULL);
dengyihao's avatar
dengyihao 已提交
195 196 197 198 199
  while (connList != NULL) {
    while (!QUEUE_IS_EMPTY(&connList->conn)) {
      queue* h = QUEUE_HEAD(&connList->conn);
      QUEUE_REMOVE(h);
      SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
200
      clientConnDestroy(c, true);
dengyihao's avatar
dengyihao 已提交
201
    }
dengyihao's avatar
dengyihao 已提交
202
    connList = taosHashIterate((SHashObj*)pool, connList);
dengyihao's avatar
dengyihao 已提交
203
  }
dengyihao's avatar
dengyihao 已提交
204
  taosHashClear(pool);
dengyihao's avatar
dengyihao 已提交
205 206
}

dengyihao's avatar
dengyihao 已提交
207
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
dengyihao's avatar
dengyihao 已提交
208 209 210 211
  char key[128] = {0};
  tstrncpy(key, ip, strlen(ip));
  tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));

dengyihao's avatar
dengyihao 已提交
212 213
  SHashObj*  pPool = pool;
  SConnList* plist = taosHashGet(pPool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
214 215
  if (plist == NULL) {
    SConnList list;
dengyihao's avatar
dengyihao 已提交
216 217
    taosHashPut(pPool, key, strlen(key), (void*)&list, sizeof(list));
    plist = taosHashGet(pPool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
218
    QUEUE_INIT(&plist->conn);
dengyihao's avatar
dengyihao 已提交
219 220 221 222 223 224 225
  }

  if (QUEUE_IS_EMPTY(&plist->conn)) {
    return NULL;
  }
  queue* h = QUEUE_HEAD(&plist->conn);
  QUEUE_REMOVE(h);
dengyihao's avatar
dengyihao 已提交
226 227 228 229

  SCliConn* conn = QUEUE_DATA(h, SCliConn, conn);
  QUEUE_INIT(&conn->conn);
  return conn;
dengyihao's avatar
dengyihao 已提交
230
}
dengyihao's avatar
dengyihao 已提交
231
static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
232
  char key[128] = {0};
dengyihao's avatar
dengyihao 已提交
233

dengyihao's avatar
dengyihao 已提交
234 235
  tstrncpy(key, ip, strlen(ip));
  tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
dengyihao's avatar
dengyihao 已提交
236
  tDebug("conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap);
dengyihao's avatar
dengyihao 已提交
237

dengyihao's avatar
dengyihao 已提交
238
  SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
dengyihao's avatar
dengyihao 已提交
239
  conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
dengyihao's avatar
dengyihao 已提交
240
  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
241
  conn->notifyCount = 0;
dengyihao's avatar
dengyihao 已提交
242 243 244 245
  // list already create before
  assert(plist != NULL);
  QUEUE_PUSH(&plist->conn, &conn->conn);
}
dengyihao's avatar
dengyihao 已提交
246 247 248 249 250 251 252 253 254
static bool clientReadComplete(SConnBuffer* data) {
  STransMsgHead head;
  int32_t       headLen = sizeof(head);
  if (data->len >= headLen) {
    memcpy((char*)&head, data->buf, headLen);
    int32_t msgLen = (int32_t)htonl((uint32_t)head.msgLen);
    if (msgLen > data->len) {
      data->left = msgLen - data->len;
      return false;
dengyihao's avatar
dengyihao 已提交
255 256
    } else if (msgLen == data->len) {
      data->left = 0;
dengyihao's avatar
dengyihao 已提交
257 258 259 260 261 262
      return true;
    }
  } else {
    return false;
  }
}
dengyihao's avatar
dengyihao 已提交
263
static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
264 265
  SCliConn*    conn = handle->data;
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
266
  transAllocBuffer(pBuf, buf);
dengyihao's avatar
dengyihao 已提交
267 268
}
static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
269
  // impl later
dengyihao's avatar
dengyihao 已提交
270 271
  SCliConn*    conn = handle->data;
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
272
  if (nread > 0) {
dengyihao's avatar
dengyihao 已提交
273 274
    pBuf->len += nread;
    if (clientReadComplete(pBuf)) {
dengyihao's avatar
dengyihao 已提交
275
      uv_read_stop((uv_stream_t*)conn->stream);
276
      tDebug("conn %p read complete", conn);
277
      clientHandleResp(conn);
dengyihao's avatar
dengyihao 已提交
278
    } else {
279
      tDebug("conn %p read partial packet, continue to read", conn);
dengyihao's avatar
dengyihao 已提交
280
    }
dengyihao's avatar
dengyihao 已提交
281 282
    return;
  }
283 284 285 286
  assert(nread <= 0);
  if (nread == 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
287
  if (nread < 0 || nread == UV_EOF) {
288
    tError("conn %p read error: %s", conn, uv_err_name(nread));
dengyihao's avatar
dengyihao 已提交
289
    clientHandleExcept(conn);
dengyihao's avatar
dengyihao 已提交
290
  }
291 292
  // tDebug("Read error %s\n", uv_err_name(nread));
  // uv_close((uv_handle_t*)handle, clientDestroy);
dengyihao's avatar
dengyihao 已提交
293
}
dengyihao's avatar
dengyihao 已提交
294

295
static void clientConnDestroy(SCliConn* conn, bool clear) {
dengyihao's avatar
dengyihao 已提交
296
  //
dengyihao's avatar
dengyihao 已提交
297 298 299 300 301 302 303 304
  conn->ref--;
  if (conn->ref == 0) {
    tDebug("conn %p remove from conn pool", conn);
    QUEUE_REMOVE(&conn->conn);
    tDebug("conn %p remove from conn pool successfully", conn);
    if (clear) {
      uv_close((uv_handle_t*)conn->stream, clientDestroy);
    }
305
  }
dengyihao's avatar
dengyihao 已提交
306
}
dengyihao's avatar
dengyihao 已提交
307 308
static void clientDestroy(uv_handle_t* handle) {
  SCliConn* conn = handle->data;
dengyihao's avatar
dengyihao 已提交
309
  // transDestroyBuffer(&conn->readBuf);
dengyihao's avatar
dengyihao 已提交
310 311 312 313 314 315 316

  free(conn->stream);
  free(conn->writeReq);
  tDebug("conn %p destroy successfully", conn);
  free(conn);

  // clientConnDestroy(conn, false);
dengyihao's avatar
dengyihao 已提交
317 318 319 320 321
}

static void clientWriteCb(uv_write_t* req, int status) {
  SCliConn* pConn = req->data;
  if (status == 0) {
322
    tDebug("conn %p data already was written out", pConn);
dengyihao's avatar
dengyihao 已提交
323 324 325 326 327 328
    SCliMsg* pMsg = pConn->data;
    if (pMsg != NULL) {
      transFreeMsg((pMsg->msg.pCont));
      pMsg->msg.pCont = NULL;
    }

dengyihao's avatar
dengyihao 已提交
329
  } else {
330
    tError("conn %p failed to write: %s", pConn, uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
331
    clientHandleExcept(pConn);
dengyihao's avatar
dengyihao 已提交
332 333
    return;
  }
dengyihao's avatar
dengyihao 已提交
334
  SCliThrdObj* pThrd = pConn->hostThrd;
dengyihao's avatar
dengyihao 已提交
335
  uv_read_start((uv_stream_t*)pConn->stream, clientAllocBufferCb, clientReadCb);
dengyihao's avatar
dengyihao 已提交
336
}
dengyihao's avatar
dengyihao 已提交
337 338

static void clientWrite(SCliConn* pConn) {
dengyihao's avatar
dengyihao 已提交
339 340 341 342
  SCliMsg*       pCliMsg = pConn->data;
  SRpcMsg*       pMsg = (SRpcMsg*)(&pCliMsg->msg);
  STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);

dengyihao's avatar
dengyihao 已提交
343
  int msgLen = transMsgLenFromCont(pMsg->contLen);
dengyihao's avatar
dengyihao 已提交
344

dengyihao's avatar
dengyihao 已提交
345 346 347 348
  pHead->msgType = pMsg->msgType;
  pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);

  uv_buf_t wb = uv_buf_init((char*)pHead, msgLen);
349
  tDebug("conn %p data write out, msgType : %d, len: %d", pConn, pHead->msgType, msgLen);
dengyihao's avatar
dengyihao 已提交
350 351
  uv_write(pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb);
}
dengyihao's avatar
dengyihao 已提交
352 353
static void clientConnCb(uv_connect_t* req, int status) {
  // impl later
dengyihao's avatar
dengyihao 已提交
354 355
  SCliConn* pConn = req->data;
  if (status != 0) {
dengyihao's avatar
dengyihao 已提交
356
    // tError("failed to connect server(%s, %d), errmsg: %s", pCtx->ip, pCtx->port, uv_strerror(status));
357
    tError("conn %p failed to connect server: %s", pConn, uv_strerror(status));
dengyihao's avatar
dengyihao 已提交
358
    clientHandleExcept(pConn);
359
    return;
dengyihao's avatar
dengyihao 已提交
360
  }
361
  tDebug("conn %p create", pConn);
dengyihao's avatar
dengyihao 已提交
362

dengyihao's avatar
dengyihao 已提交
363
  assert(pConn->stream == req->handle);
dengyihao's avatar
dengyihao 已提交
364
  clientWrite(pConn);
dengyihao's avatar
dengyihao 已提交
365 366
}

dengyihao's avatar
dengyihao 已提交
367
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
dengyihao's avatar
dengyihao 已提交
368 369
  uint64_t et = taosGetTimestampUs();
  uint64_t el = et - pMsg->st;
dengyihao's avatar
dengyihao 已提交
370
  tDebug("msg tran time cost: %" PRIu64 "", el);
dengyihao's avatar
dengyihao 已提交
371
  et = taosGetTimestampUs();
dengyihao's avatar
dengyihao 已提交
372

dengyihao's avatar
dengyihao 已提交
373
  STransConnCtx* pCtx = pMsg->ctx;
dengyihao's avatar
dengyihao 已提交
374
  SCliConn*      conn = getConnFromPool(pThrd->pool, pCtx->ip, pCtx->port);
dengyihao's avatar
dengyihao 已提交
375
  if (conn != NULL) {
dengyihao's avatar
dengyihao 已提交
376
    // impl later
377
    tDebug("conn %p get from conn pool", conn);
dengyihao's avatar
dengyihao 已提交
378 379
    conn->data = pMsg;
    conn->writeReq->data = conn;
dengyihao's avatar
dengyihao 已提交
380
    transDestroyBuffer(&conn->readBuf);
dengyihao's avatar
dengyihao 已提交
381
    clientWrite(conn);
dengyihao's avatar
dengyihao 已提交
382
  } else {
dengyihao's avatar
dengyihao 已提交
383
    SCliConn* conn = calloc(1, sizeof(SCliConn));
dengyihao's avatar
dengyihao 已提交
384
    conn->ref++;
dengyihao's avatar
dengyihao 已提交
385
    // read/write stream handle
dengyihao's avatar
dengyihao 已提交
386 387
    conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
    uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
dengyihao's avatar
dengyihao 已提交
388 389 390
    conn->stream->data = conn;

    // write req handle
dengyihao's avatar
dengyihao 已提交
391
    conn->writeReq = malloc(sizeof(uv_write_t));
dengyihao's avatar
dengyihao 已提交
392
    conn->writeReq->data = conn;
dengyihao's avatar
dengyihao 已提交
393

dengyihao's avatar
dengyihao 已提交
394
    QUEUE_INIT(&conn->conn);
dengyihao's avatar
dengyihao 已提交
395 396 397

    conn->connReq.data = conn;
    conn->data = pMsg;
dengyihao's avatar
dengyihao 已提交
398
    conn->hostThrd = pThrd;
dengyihao's avatar
dengyihao 已提交
399

dengyihao's avatar
dengyihao 已提交
400
    struct sockaddr_in addr;
dengyihao's avatar
dengyihao 已提交
401
    uv_ip4_addr(pMsg->ctx->ip, pMsg->ctx->port, &addr);
dengyihao's avatar
dengyihao 已提交
402
    // handle error in callback if fail to connect
dengyihao's avatar
dengyihao 已提交
403
    uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, clientConnCb);
dengyihao's avatar
dengyihao 已提交
404 405 406 407 408 409
  }
}
static void clientAsyncCb(uv_async_t* handle) {
  SCliThrdObj* pThrd = handle->data;
  SCliMsg*     pMsg = NULL;
  queue        wq;
dengyihao's avatar
dengyihao 已提交
410

dengyihao's avatar
dengyihao 已提交
411 412 413 414
  // batch process to avoid to lock/unlock frequently
  pthread_mutex_lock(&pThrd->msgMtx);
  QUEUE_MOVE(&pThrd->msg, &wq);
  pthread_mutex_unlock(&pThrd->msgMtx);
dengyihao's avatar
dengyihao 已提交
415

dengyihao's avatar
dengyihao 已提交
416 417 418 419
  int count = 0;
  while (!QUEUE_IS_EMPTY(&wq)) {
    queue* h = QUEUE_HEAD(&wq);
    QUEUE_REMOVE(h);
dengyihao's avatar
dengyihao 已提交
420 421

    SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
dengyihao's avatar
dengyihao 已提交
422 423
    clientHandleReq(pMsg, pThrd);
    count++;
dengyihao's avatar
dengyihao 已提交
424 425 426
  }
  if (count >= 2) {
    tDebug("already process batch size: %d", count);
dengyihao's avatar
dengyihao 已提交
427
  }
dengyihao's avatar
dengyihao 已提交
428 429 430 431 432 433 434 435 436
}

static void* clientThread(void* arg) {
  SCliThrdObj* pThrd = (SCliThrdObj*)arg;
  uv_run(pThrd->loop, UV_RUN_DEFAULT);
}

void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
  SClientObj* cli = calloc(1, sizeof(SClientObj));
dengyihao's avatar
dengyihao 已提交
437

dengyihao's avatar
dengyihao 已提交
438
  SRpcInfo* pRpc = shandle;
dengyihao's avatar
dengyihao 已提交
439 440 441 442 443
  memcpy(cli->label, label, strlen(label));
  cli->numOfThreads = numOfThreads;
  cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*));

  for (int i = 0; i < cli->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
444
    SCliThrdObj* pThrd = createThrdObj();
dengyihao's avatar
dengyihao 已提交
445
    pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
dengyihao's avatar
dengyihao 已提交
446
    pThrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
447

dengyihao's avatar
dengyihao 已提交
448
    int err = pthread_create(&pThrd->thread, NULL, clientThread, (void*)(pThrd));
dengyihao's avatar
dengyihao 已提交
449 450 451
    if (err == 0) {
      tDebug("sucess to create tranport-client thread %d", i);
    }
dengyihao's avatar
dengyihao 已提交
452
    cli->pThreadObj[i] = pThrd;
dengyihao's avatar
dengyihao 已提交
453 454 455
  }
  return cli;
}
dengyihao's avatar
dengyihao 已提交
456 457 458 459
static void clientMsgDestroy(SCliMsg* pMsg) {
  // impl later
  free(pMsg);
}
dengyihao's avatar
dengyihao 已提交
460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475
static SCliThrdObj* createThrdObj() {
  SCliThrdObj* pThrd = (SCliThrdObj*)calloc(1, sizeof(SCliThrdObj));
  QUEUE_INIT(&pThrd->msg);
  pthread_mutex_init(&pThrd->msgMtx, NULL);

  pThrd->loop = (uv_loop_t*)malloc(sizeof(uv_loop_t));
  uv_loop_init(pThrd->loop);

  pThrd->cliAsync = malloc(sizeof(uv_async_t));
  uv_async_init(pThrd->loop, pThrd->cliAsync, clientAsyncCb);
  pThrd->cliAsync->data = pThrd;

  pThrd->pTimer = malloc(sizeof(uv_timer_t));
  uv_timer_init(pThrd->loop, pThrd->pTimer);
  pThrd->pTimer->data = pThrd;

476
  pThrd->pool = creatConnPool(1);
dengyihao's avatar
dengyihao 已提交
477 478
  return pThrd;
}
dengyihao's avatar
dengyihao 已提交
479
static void destroyThrdObj(SCliThrdObj* pThrd) {
dengyihao's avatar
dengyihao 已提交
480 481 482
  if (pThrd == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
483 484 485 486 487 488
  pthread_join(pThrd->thread, NULL);
  pthread_mutex_destroy(&pThrd->msgMtx);
  free(pThrd->cliAsync);
  free(pThrd->loop);
  free(pThrd);
}
dengyihao's avatar
dengyihao 已提交
489 490 491 492 493 494 495

static void destroyTransConnCtx(STransConnCtx* ctx) {
  if (ctx != NULL) {
    free(ctx->ip);
  }
  free(ctx);
}
dengyihao's avatar
dengyihao 已提交
496
//
dengyihao's avatar
dengyihao 已提交
497 498 499
void taosCloseClient(void* arg) {
  // impl later
  SClientObj* cli = arg;
dengyihao's avatar
dengyihao 已提交
500
  for (int i = 0; i < cli->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
501
    destroyThrdObj(cli->pThreadObj[i]);
dengyihao's avatar
dengyihao 已提交
502 503 504
  }
  free(cli->pThreadObj);
  free(cli);
dengyihao's avatar
dengyihao 已提交
505
}
dengyihao's avatar
dengyihao 已提交
506 507
void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) {
  // impl later
dengyihao's avatar
dengyihao 已提交
508 509 510
  char*    ip = (char*)(pEpSet->fqdn[pEpSet->inUse]);
  uint32_t port = pEpSet->port[pEpSet->inUse];

dengyihao's avatar
dengyihao 已提交
511 512
  SRpcInfo* pRpc = (SRpcInfo*)shandle;

dengyihao's avatar
dengyihao 已提交
513 514 515 516
  int32_t flen = 0;
  if (transCompressMsg(pMsg->pCont, pMsg->contLen, &flen)) {
    // imp later
  }
dengyihao's avatar
dengyihao 已提交
517

dengyihao's avatar
dengyihao 已提交
518 519
  STransConnCtx* pCtx = calloc(1, sizeof(STransConnCtx));

dengyihao's avatar
dengyihao 已提交
520
  pCtx->pTransInst = (SRpcInfo*)shandle;
dengyihao's avatar
dengyihao 已提交
521 522 523 524
  pCtx->ahandle = pMsg->ahandle;
  pCtx->msgType = pMsg->msgType;
  pCtx->ip = strdup(ip);
  pCtx->port = port;
dengyihao's avatar
dengyihao 已提交
525 526 527 528 529 530 531

  assert(pRpc->connType == TAOS_CONN_CLIENT);
  // atomic or not
  int64_t index = pRpc->index;
  if (pRpc->index++ >= pRpc->numOfThreads) {
    pRpc->index = 0;
  }
dengyihao's avatar
dengyihao 已提交
532 533 534 535
  SCliMsg* cliMsg = malloc(sizeof(SCliMsg));
  cliMsg->ctx = pCtx;
  cliMsg->msg = *pMsg;
  cliMsg->st = taosGetTimestampUs();
dengyihao's avatar
dengyihao 已提交
536 537 538 539

  SCliThrdObj* thrd = ((SClientObj*)pRpc->tcphandle)->pThreadObj[index % pRpc->numOfThreads];

  pthread_mutex_lock(&thrd->msgMtx);
dengyihao's avatar
dengyihao 已提交
540
  QUEUE_PUSH(&thrd->msg, &cliMsg->q);
dengyihao's avatar
dengyihao 已提交
541 542 543 544 545
  pthread_mutex_unlock(&thrd->msgMtx);

  uv_async_send(thrd->cliAsync);
}
#endif