transCli.c 38.8 KB
Newer Older
dengyihao's avatar
dengyihao 已提交
1
/* * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
dengyihao's avatar
dengyihao 已提交
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
 *
 * 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"

dengyihao's avatar
dengyihao 已提交
18 19 20
static int32_t transSCliInst = 0;
static int32_t refMgt = 0;

dengyihao's avatar
dengyihao 已提交
21
typedef struct SCliConn {
dengyihao's avatar
dengyihao 已提交
22
  T_REF_DECLARE()
dengyihao's avatar
dengyihao 已提交
23 24
  uv_connect_t connReq;
  uv_stream_t* stream;
dengyihao's avatar
dengyihao 已提交
25
  uv_write_t   writeReq;
dengyihao's avatar
dengyihao 已提交
26

dengyihao's avatar
dengyihao 已提交
27 28 29 30 31 32 33 34 35
  void* hostThrd;
  int   hThrdIdx;

  SConnBuffer readBuf;
  STransQueue cliMsgs;
  queue       conn;
  uint64_t    expireTime;

  STransCtx  ctx;
dengyihao's avatar
dengyihao 已提交
36 37
  bool       broken;  // link broken or not
  ConnStatus status;  //
dengyihao's avatar
dengyihao 已提交
38 39 40 41

  char*    ip;
  uint32_t port;

dengyihao's avatar
dengyihao 已提交
42
  // debug and log info
dengyihao's avatar
dengyihao 已提交
43
  struct sockaddr_in addr;
dengyihao's avatar
dengyihao 已提交
44
  struct sockaddr_in localAddr;
dengyihao's avatar
dengyihao 已提交
45
} SCliConn;
dengyihao's avatar
dengyihao 已提交
46

dengyihao's avatar
dengyihao 已提交
47
typedef struct SCliMsg {
dengyihao's avatar
dengyihao 已提交
48
  STransConnCtx* ctx;
dengyihao's avatar
formate  
dengyihao 已提交
49
  STransMsg      msg;
dengyihao's avatar
dengyihao 已提交
50 51
  queue          q;
  uint64_t       st;
dengyihao's avatar
dengyihao 已提交
52
  STransMsgType  type;
dengyihao's avatar
dengyihao 已提交
53
  int            sent;  //(0: no send, 1: alread sent)
dengyihao's avatar
dengyihao 已提交
54 55 56
} SCliMsg;

typedef struct SCliThrdObj {
dengyihao's avatar
dengyihao 已提交
57 58
  TdThread    thread;  // tid
  int64_t     pid;     // pid
dengyihao's avatar
dengyihao 已提交
59
  uv_loop_t*  loop;
dengyihao's avatar
dengyihao 已提交
60
  SAsyncPool* asyncPool;
dengyihao's avatar
dengyihao 已提交
61
  uv_timer_t  timer;
dengyihao's avatar
dengyihao 已提交
62 63 64
  void*       pool;  // conn pool

  // msg queue
dengyihao's avatar
dengyihao 已提交
65
  queue         msg;
wafwerar's avatar
wafwerar 已提交
66
  TdThreadMutex msgMtx;
dengyihao's avatar
dengyihao 已提交
67 68 69
  SDelayQueue*  delayQueue;
  uint64_t      nextTimeout;  // next timeout
  void*         pTransInst;   //
dengyihao's avatar
dengyihao 已提交
70

dengyihao's avatar
dengyihao 已提交
71
  SCvtAddr cvtAddr;
dengyihao's avatar
dengyihao 已提交
72 73

  bool quit;
dengyihao's avatar
dengyihao 已提交
74 75
} SCliThrdObj;

U
ubuntu 已提交
76
typedef struct SCliObj {
dengyihao's avatar
dengyihao 已提交
77 78 79 80
  char          label[TSDB_LABEL_LEN];
  int32_t       index;
  int           numOfThreads;
  SCliThrdObj** pThreadObj;
U
ubuntu 已提交
81
} SCliObj;
dengyihao's avatar
dengyihao 已提交
82

dengyihao's avatar
dengyihao 已提交
83 84 85 86
typedef struct SConnList {
  queue conn;
} SConnList;

dengyihao's avatar
dengyihao 已提交
87
// conn pool
dengyihao's avatar
dengyihao 已提交
88
// add expire timeout and capacity limit
dengyihao's avatar
dengyihao 已提交
89
static void*     createConnPool(int size);
90
static void*     destroyConnPool(void* pool);
dengyihao's avatar
dengyihao 已提交
91
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port);
dengyihao's avatar
dengyihao 已提交
92
static void      addConnToPool(void* pool, SCliConn* conn);
dengyihao's avatar
dengyihao 已提交
93

dengyihao's avatar
dengyihao 已提交
94
// register timer in each thread to clear expire conn
U
ubuntu 已提交
95 96
static void cliTimeoutCb(uv_timer_t* handle);
// alloc buf for recv
dengyihao's avatar
dengyihao 已提交
97
static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
U
ubuntu 已提交
98 99
// callback after  read nbytes from socket
static void cliRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
100
// callback after write data to socket
U
ubuntu 已提交
101
static void cliSendCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
102
// callback after conn  to server
U
ubuntu 已提交
103 104
static void cliConnCb(uv_connect_t* req, int status);
static void cliAsyncCb(uv_async_t* handle);
dengyihao's avatar
dengyihao 已提交
105

dengyihao's avatar
dengyihao 已提交
106
static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg);
dengyihao's avatar
dengyihao 已提交
107

U
ubuntu 已提交
108 109 110
static SCliConn* cliCreateConn(SCliThrdObj* thrd);
static void      cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
static void      cliDestroy(uv_handle_t* handle);
dengyihao's avatar
dengyihao 已提交
111
static void      cliSend(SCliConn* pConn);
dengyihao's avatar
dengyihao 已提交
112

dengyihao's avatar
dengyihao 已提交
113
void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr);
dengyihao's avatar
dengyihao 已提交
114 115 116 117
/*
 * set TCP connection timeout per-socket level
 */
static int cliCreateSocket();
dengyihao's avatar
dengyihao 已提交
118
// process data read from server, add decompress etc later
U
ubuntu 已提交
119
static void cliHandleResp(SCliConn* conn);
dengyihao's avatar
dengyihao 已提交
120
// handle except about conn
U
ubuntu 已提交
121
static void cliHandleExcept(SCliConn* conn);
dengyihao's avatar
dengyihao 已提交
122

123
// handle req from app
U
ubuntu 已提交
124 125
static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd);
dengyihao's avatar
dengyihao 已提交
126
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd);
dengyihao's avatar
dengyihao 已提交
127 128
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd);
static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrdObj* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease,
dengyihao's avatar
dengyihao 已提交
129
                                                                      NULL, cliHandleUpdate};
dengyihao's avatar
dengyihao 已提交
130

U
ubuntu 已提交
131
static void cliSendQuit(SCliThrdObj* thrd);
U
ubuntu 已提交
132
static void destroyUserdata(STransMsg* userdata);
dengyihao's avatar
dengyihao 已提交
133

U
ubuntu 已提交
134
static int cliRBChoseIdx(STrans* pTransInst);
dengyihao's avatar
dengyihao 已提交
135

dengyihao's avatar
dengyihao 已提交
136 137
static void destroyCmsg(SCliMsg* cmsg);
static void transDestroyConnCtx(STransConnCtx* ctx);
dengyihao's avatar
dengyihao 已提交
138 139 140
// thread obj
static SCliThrdObj* createThrdObj();
static void         destroyThrdObj(SCliThrdObj* pThrd);
dengyihao's avatar
dengyihao 已提交
141

dengyihao's avatar
dengyihao 已提交
142 143
static void cliWalkCb(uv_handle_t* handle, void* arg);

dengyihao's avatar
dengyihao 已提交
144 145 146 147 148 149 150 151 152 153 154 155 156
static void cliReleaseUnfinishedMsg(SCliConn* conn) {
  SCliMsg* pMsg = NULL;
  for (int i = 0; i < transQueueSize(&conn->cliMsgs); i++) {
    pMsg = transQueueGet(&conn->cliMsgs, i);
    if (pMsg != NULL && pMsg->ctx != NULL) {
      if (conn->ctx.freeFunc != NULL) {
        conn->ctx.freeFunc(pMsg->ctx->ahandle);
      }
    }
    destroyCmsg(pMsg);
  }
}

dengyihao's avatar
dengyihao 已提交
157 158 159 160 161 162 163
#define CLI_RELEASE_UV(loop)        \
  do {                              \
    uv_walk(loop, cliWalkCb, NULL); \
    uv_run(loop, UV_RUN_DEFAULT);   \
    uv_loop_close(loop);            \
  } while (0);

dengyihao's avatar
dengyihao 已提交
164 165 166 167 168 169
// snprintf may cause performance problem
#define CONN_CONSTRUCT_HASH_KEY(key, ip, port)          \
  do {                                                  \
    snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \
  } while (0)

dengyihao's avatar
dengyihao 已提交
170
#define CONN_HOST_THREAD_INDEX(conn) (conn ? ((SCliConn*)conn)->hThrdIdx : -1)
dengyihao's avatar
dengyihao 已提交
171 172 173
#define CONN_PERSIST_TIME(para)      (para * 1000 * 10)
#define CONN_GET_HOST_THREAD(conn)   (conn ? ((SCliConn*)conn)->hostThrd : NULL)
#define CONN_GET_INST_LABEL(conn)    (((STrans*)(((SCliThrdObj*)(conn)->hostThrd)->pTransInst))->label)
dengyihao's avatar
dengyihao 已提交
174 175 176
#define CONN_SHOULD_RELEASE(conn, head)                                                  \
  do {                                                                                   \
    if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) {                       \
dengyihao's avatar
dengyihao 已提交
177 178
      uint64_t ahandle = head->ahandle;                                                  \
      CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle);                                         \
dengyihao's avatar
dengyihao 已提交
179 180 181 182
      conn->status = ConnRelease;                                                        \
      transClearBuffer(&conn->readBuf);                                                  \
      transFreeMsg(transContFromHead((char*)head));                                      \
      tDebug("cli conn %p receive release request, ref: %d", conn, T_REF_VAL_GET(conn)); \
dengyihao's avatar
dengyihao 已提交
183
      if (T_REF_VAL_GET(conn) > 1) {                                                     \
184
        transUnrefCliHandle(conn);                                                       \
dengyihao's avatar
dengyihao 已提交
185
      }                                                                                  \
dengyihao's avatar
dengyihao 已提交
186
      destroyCmsg(pMsg);                                                                 \
dengyihao's avatar
dengyihao 已提交
187
      cliReleaseUnfinishedMsg(conn);                                                     \
dengyihao's avatar
dengyihao 已提交
188
      addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn);                         \
dengyihao's avatar
dengyihao 已提交
189 190
      return;                                                                            \
    }                                                                                    \
dengyihao's avatar
dengyihao 已提交
191 192
  } while (0)

dengyihao's avatar
dengyihao 已提交
193 194 195 196 197
#define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle)                                         \
  do {                                                                                    \
    int i = 0, sz = transQueueSize(&conn->cliMsgs);                                       \
    for (; i < sz; i++) {                                                                 \
      pMsg = transQueueGet(&conn->cliMsgs, i);                                            \
S
Shengliang 已提交
198
      if (pMsg != NULL && pMsg->ctx != NULL && (uint64_t)pMsg->ctx->ahandle == ahandle) { \
dengyihao's avatar
dengyihao 已提交
199 200 201 202 203 204 205 206
        break;                                                                            \
      }                                                                                   \
    }                                                                                     \
    if (i == sz) {                                                                        \
      pMsg = NULL;                                                                        \
    } else {                                                                              \
      pMsg = transQueueRm(&conn->cliMsgs, i);                                             \
    }                                                                                     \
dengyihao's avatar
dengyihao 已提交
207
  } while (0)
U
ubuntu 已提交
208 209 210 211 212 213 214 215 216 217 218 219 220
#define CONN_GET_NEXT_SENDMSG(conn)                 \
  do {                                              \
    int i = 0;                                      \
    do {                                            \
      pCliMsg = transQueueGet(&conn->cliMsgs, i++); \
      if (pCliMsg && 0 == pCliMsg->sent) {          \
        break;                                      \
      }                                             \
    } while (pCliMsg != NULL);                      \
    if (pCliMsg == NULL) {                          \
      goto _RETURN;                                 \
    }                                               \
  } while (0)
dengyihao's avatar
dengyihao 已提交
221

dengyihao's avatar
dengyihao 已提交
222 223 224 225 226
#define CONN_HANDLE_THREAD_QUIT(thrd) \
  do {                                \
    if (thrd->quit) {                 \
      return;                         \
    }                                 \
dengyihao's avatar
dengyihao 已提交
227 228 229 230 231
  } while (0)

#define CONN_HANDLE_BROKEN(conn) \
  do {                           \
    if (conn->broken) {          \
dengyihao's avatar
formate  
dengyihao 已提交
232
      cliHandleExcept(conn);     \
dengyihao's avatar
dengyihao 已提交
233
      return;                    \
dengyihao's avatar
dengyihao 已提交
234
    }                            \
dengyihao's avatar
dengyihao 已提交
235
  } while (0)
dengyihao's avatar
dengyihao 已提交
236

dengyihao's avatar
formate  
dengyihao 已提交
237 238
#define CONN_SET_PERSIST_BY_APP(conn) \
  do {                                \
dengyihao's avatar
dengyihao 已提交
239 240
    if (conn->status == ConnNormal) { \
      conn->status = ConnAcquire;     \
dengyihao's avatar
formate  
dengyihao 已提交
241 242 243
      transRefCliHandle(conn);        \
    }                                 \
  } while (0)
244

dengyihao's avatar
dengyihao 已提交
245 246 247 248
#define CONN_NO_PERSIST_BY_APP(conn) \
  (((conn)->status == ConnNormal || (conn)->status == ConnInPool) && T_REF_VAL_GET(conn) == 1)
#define CONN_RELEASE_BY_SERVER(conn) \
  (((conn)->status == ConnRelease || (conn)->status == ConnInPool) && T_REF_VAL_GET(conn) == 1)
249

S
Shengliang Guan 已提交
250 251
#define REQUEST_NO_RESP(msg)         ((msg)->info.noResp == 1)
#define REQUEST_PERSIS_HANDLE(msg)   ((msg)->info.persistHandle == 1)
dengyihao's avatar
dengyihao 已提交
252
#define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release)
dengyihao's avatar
dengyihao 已提交
253

dengyihao's avatar
dengyihao 已提交
254
#define EPSET_GET_INUSE_IP(epSet)   ((epSet)->eps[(epSet)->inUse].fqdn)
dengyihao's avatar
dengyihao 已提交
255 256
#define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port)

U
ubuntu 已提交
257
static void* cliWorkThread(void* arg);
dengyihao's avatar
dengyihao 已提交
258

dengyihao's avatar
dengyihao 已提交
259
bool cliMaySendCachedMsg(SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
260
  if (!transQueueEmpty(&conn->cliMsgs)) {
dengyihao's avatar
dengyihao 已提交
261
    SCliMsg* pCliMsg = NULL;
U
ubuntu 已提交
262
    CONN_GET_NEXT_SENDMSG(conn);
dengyihao's avatar
dengyihao 已提交
263 264
    cliSend(conn);
  }
dengyihao's avatar
dengyihao 已提交
265
  return false;
U
ubuntu 已提交
266 267
_RETURN:
  return false;
dengyihao's avatar
dengyihao 已提交
268
}
dengyihao's avatar
formate  
dengyihao 已提交
269
void cliHandleResp(SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
270
  SCliThrdObj* pThrd = conn->hostThrd;
dengyihao's avatar
formate  
dengyihao 已提交
271
  STrans*      pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
272

dengyihao's avatar
dengyihao 已提交
273 274 275
  STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
  pHead->code = htonl(pHead->code);
  pHead->msgLen = htonl(pHead->msgLen);
dengyihao's avatar
dengyihao 已提交
276

dengyihao's avatar
dengyihao 已提交
277 278 279 280 281
  STransMsg transMsg = {0};
  transMsg.contLen = transContLenFromMsg(pHead->msgLen);
  transMsg.pCont = transContFromHead((char*)pHead);
  transMsg.code = pHead->code;
  transMsg.msgType = pHead->msgType;
S
Shengliang Guan 已提交
282
  transMsg.info.ahandle = NULL;
dengyihao's avatar
dengyihao 已提交
283

dengyihao's avatar
dengyihao 已提交
284 285
  SCliMsg*       pMsg = NULL;
  STransConnCtx* pCtx = NULL;
dengyihao's avatar
dengyihao 已提交
286
  CONN_SHOULD_RELEASE(conn, pHead);
dengyihao's avatar
dengyihao 已提交
287

dengyihao's avatar
dengyihao 已提交
288 289
  if (CONN_NO_PERSIST_BY_APP(conn)) {
    pMsg = transQueuePop(&conn->cliMsgs);
dengyihao's avatar
dengyihao 已提交
290 291
    pCtx = pMsg ? pMsg->ctx : NULL;
    if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) {
S
Shengliang Guan 已提交
292 293 294
      transMsg.info.ahandle = transCtxDumpVal(&conn->ctx, transMsg.msgType);
      if (transMsg.info.ahandle == NULL) {
        transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&conn->ctx, (int32_t*)&(transMsg.msgType));
dengyihao's avatar
dengyihao 已提交
295
      }
S
Shengliang Guan 已提交
296
      tDebug("cli conn %p construct ahandle %p, persist: 0", conn, transMsg.info.ahandle);
dengyihao's avatar
dengyihao 已提交
297
    } else {
S
Shengliang Guan 已提交
298 299
      transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL;
      tDebug("cli conn %p get ahandle %p, persist: 0", conn, transMsg.info.ahandle);
dengyihao's avatar
dengyihao 已提交
300
    }
U
ubuntu 已提交
301
  } else {
dengyihao's avatar
dengyihao 已提交
302 303 304
    uint64_t ahandle = (uint64_t)pHead->ahandle;
    CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle);
    if (pMsg == NULL) {
S
Shengliang Guan 已提交
305 306 307 308
      transMsg.info.ahandle = transCtxDumpVal(&conn->ctx, transMsg.msgType);
      tDebug("cli conn %p construct ahandle %p by %s, persist: 1", conn, transMsg.info.ahandle,
             TMSG_INFO(transMsg.msgType));
      if (!CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) {
dengyihao's avatar
dengyihao 已提交
309
        transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
S
Shengliang Guan 已提交
310 311
        transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&conn->ctx, (int32_t*)&(transMsg.msgType));
        tDebug("cli conn %p construct ahandle %p due brokenlink, persist: 1", conn, transMsg.info.ahandle);
dengyihao's avatar
dengyihao 已提交
312 313 314
      }
    } else {
      pCtx = pMsg ? pMsg->ctx : NULL;
S
Shengliang Guan 已提交
315 316
      transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL;
      tDebug("cli conn %p get ahandle %p, persist: 1", conn, transMsg.info.ahandle);
dengyihao's avatar
dengyihao 已提交
317
    }
U
ubuntu 已提交
318
  }
dengyihao's avatar
dengyihao 已提交
319 320 321
  // buf's mem alread translated to transMsg.pCont
  transClearBuffer(&conn->readBuf);

dengyihao's avatar
dengyihao 已提交
322
  if (!CONN_NO_PERSIST_BY_APP(conn)) {
S
Shengliang Guan 已提交
323
    transMsg.info.handle = conn;
U
ubuntu 已提交
324
    tDebug("%s cli conn %p ref by app", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
325 326
  }

dengyihao's avatar
dengyihao 已提交
327 328
  tDebug("%s cli conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pTransInst->label, conn,
         TMSG_INFO(pHead->msgType), taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port),
dengyihao's avatar
dengyihao 已提交
329
         taosInetNtoa(conn->localAddr.sin_addr), ntohs(conn->localAddr.sin_port), transMsg.contLen);
dengyihao's avatar
dengyihao 已提交
330

dengyihao's avatar
dengyihao 已提交
331 332 333 334 335
  if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) {
    tTrace("except, server continue send while cli ignore it");
    // transUnrefCliHandle(conn);
    return;
  }
S
Shengliang Guan 已提交
336
  if (CONN_RELEASE_BY_SERVER(conn) && transMsg.info.ahandle == NULL) {
dengyihao's avatar
dengyihao 已提交
337 338 339 340
    tTrace("except, server continue send while cli ignore it");
    // transUnrefCliHandle(conn);
    return;
  }
dengyihao's avatar
dengyihao 已提交
341

dengyihao's avatar
dengyihao 已提交
342
  if (cliAppCb(conn, &transMsg, pMsg) != 0) {
dengyihao's avatar
dengyihao 已提交
343 344
    tTrace("try to send req to next node");
    return;
dengyihao's avatar
dengyihao 已提交
345
  }
dengyihao's avatar
dengyihao 已提交
346 347
  destroyCmsg(pMsg);

dengyihao's avatar
dengyihao 已提交
348
  if (cliMaySendCachedMsg(conn) == true) {
dengyihao's avatar
dengyihao 已提交
349 350
    return;
  }
dengyihao's avatar
dengyihao 已提交
351

U
ubuntu 已提交
352
  if (CONN_NO_PERSIST_BY_APP(conn)) {
dengyihao's avatar
dengyihao 已提交
353
    addConnToPool(pThrd->pool, conn);
dengyihao's avatar
dengyihao 已提交
354
  }
dengyihao's avatar
test  
dengyihao 已提交
355

dengyihao's avatar
dengyihao 已提交
356
  uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
dengyihao's avatar
dengyihao 已提交
357
  // start thread's timer of conn pool if not active
dengyihao's avatar
dengyihao 已提交
358
  if (!uv_is_active((uv_handle_t*)&pThrd->timer) && pTransInst->idleTime > 0) {
U
ubuntu 已提交
359
    // uv_timer_start((uv_timer_t*)&pThrd->timer, cliTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
dengyihao's avatar
dengyihao 已提交
360
  }
dengyihao's avatar
dengyihao 已提交
361
}
U
ubuntu 已提交
362 363

void cliHandleExcept(SCliConn* pConn) {
dengyihao's avatar
dengyihao 已提交
364
  if (transQueueEmpty(&pConn->cliMsgs)) {
dengyihao's avatar
dengyihao 已提交
365
    if (pConn->broken == true && CONN_NO_PERSIST_BY_APP(pConn)) {
dengyihao's avatar
dengyihao 已提交
366
      tTrace("%s cli conn %p handle except, persist:0", CONN_GET_INST_LABEL(pConn), pConn);
dengyihao's avatar
formate  
dengyihao 已提交
367 368 369
      transUnrefCliHandle(pConn);
      return;
    }
dengyihao's avatar
dengyihao 已提交
370
  }
dengyihao's avatar
dengyihao 已提交
371
  SCliThrdObj* pThrd = pConn->hostThrd;
dengyihao's avatar
formate  
dengyihao 已提交
372
  STrans*      pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
373
  bool         once = false;
D
dapan1121 已提交
374
  do {
dengyihao's avatar
dengyihao 已提交
375
    SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs);
D
dapan1121 已提交
376 377
    if (pMsg == NULL && once) {
      break;
dengyihao's avatar
dengyihao 已提交
378
    }
dengyihao's avatar
dengyihao 已提交
379
    STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
U
ubuntu 已提交
380

dengyihao's avatar
dengyihao 已提交
381 382 383
    STransMsg transMsg = {0};
    transMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
    transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
S
Shengliang Guan 已提交
384 385
    transMsg.info.ahandle = NULL;
    transMsg.info.handle = pConn;
dengyihao's avatar
dengyihao 已提交
386

dengyihao's avatar
dengyihao 已提交
387
    if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(pConn)) {
S
Shengliang Guan 已提交
388 389
      transMsg.info.ahandle = transCtxDumpVal(&pConn->ctx, transMsg.msgType);
      tDebug("%s cli conn %p construct ahandle %p by %s", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle,
dengyihao's avatar
dengyihao 已提交
390
             TMSG_INFO(transMsg.msgType));
S
Shengliang Guan 已提交
391 392
      if (transMsg.info.ahandle == NULL) {
        transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&pConn->ctx, (int32_t*)&(transMsg.msgType));
dengyihao's avatar
dengyihao 已提交
393
        tDebug("%s cli conn %p construct ahandle %p due to brokenlink", CONN_GET_INST_LABEL(pConn), pConn,
S
Shengliang Guan 已提交
394
               transMsg.info.ahandle);
dengyihao's avatar
dengyihao 已提交
395
      }
dengyihao's avatar
dengyihao 已提交
396
    } else {
S
Shengliang Guan 已提交
397
      transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL;
dengyihao's avatar
dengyihao 已提交
398 399 400
    }

    if (pCtx == NULL || pCtx->pSem == NULL) {
S
Shengliang Guan 已提交
401
      if (transMsg.info.ahandle == NULL) {
D
dapan1121 已提交
402
        once = true;
U
ubuntu 已提交
403 404
        continue;
      }
dengyihao's avatar
dengyihao 已提交
405
    }
dengyihao's avatar
dengyihao 已提交
406
    if (cliAppCb(pConn, &transMsg, pMsg) != 0) {
dengyihao's avatar
dengyihao 已提交
407 408
      tTrace("try to send req to next node");
      return;
dengyihao's avatar
dengyihao 已提交
409 410 411
    }
    destroyCmsg(pMsg);
    tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
D
dapan1121 已提交
412
  } while (!transQueueEmpty(&pConn->cliMsgs));
dengyihao's avatar
dengyihao 已提交
413
  transUnrefCliHandle(pConn);
dengyihao's avatar
dengyihao 已提交
414
}
dengyihao's avatar
dengyihao 已提交
415

U
ubuntu 已提交
416
void cliTimeoutCb(uv_timer_t* handle) {
dengyihao's avatar
dengyihao 已提交
417
  SCliThrdObj* pThrd = handle->data;
dengyihao's avatar
dengyihao 已提交
418
  STrans*      pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
419
  int64_t      currentTime = pThrd->nextTimeout;
dengyihao's avatar
dengyihao 已提交
420
  tTrace("%s, cli conn timeout, try to remove expire conn from conn pool", pTransInst->label);
dengyihao's avatar
dengyihao 已提交
421

dengyihao's avatar
dengyihao 已提交
422
  SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
dengyihao's avatar
dengyihao 已提交
423 424 425 426 427 428
  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);
dengyihao's avatar
dengyihao 已提交
429
        transUnrefCliHandle(c);
dengyihao's avatar
dengyihao 已提交
430 431 432 433
      } else {
        break;
      }
    }
dengyihao's avatar
dengyihao 已提交
434
    p = taosHashIterate((SHashObj*)pThrd->pool, p);
dengyihao's avatar
dengyihao 已提交
435 436
  }

dengyihao's avatar
dengyihao 已提交
437 438
  pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
  uv_timer_start(handle, cliTimeoutCb, CONN_PERSIST_TIME(pTransInst->idleTime) / 2, 0);
dengyihao's avatar
dengyihao 已提交
439
}
U
ubuntu 已提交
440 441

void* createConnPool(int size) {
442 443
  // thread local, no lock
  return taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
dengyihao's avatar
dengyihao 已提交
444
}
U
ubuntu 已提交
445
void* destroyConnPool(void* pool) {
dengyihao's avatar
dengyihao 已提交
446
  SConnList* connList = taosHashIterate((SHashObj*)pool, NULL);
dengyihao's avatar
dengyihao 已提交
447 448 449 450 451
  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);
U
ubuntu 已提交
452
      cliDestroyConn(c, true);
dengyihao's avatar
dengyihao 已提交
453
    }
dengyihao's avatar
dengyihao 已提交
454
    connList = taosHashIterate((SHashObj*)pool, connList);
dengyihao's avatar
dengyihao 已提交
455
  }
dengyihao's avatar
dengyihao 已提交
456
  taosHashCleanup(pool);
457
  return NULL;
dengyihao's avatar
dengyihao 已提交
458 459
}

dengyihao's avatar
dengyihao 已提交
460
static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
dengyihao's avatar
dengyihao 已提交
461
  char key[128] = {0};
dengyihao's avatar
dengyihao 已提交
462
  CONN_CONSTRUCT_HASH_KEY(key, ip, port);
dengyihao's avatar
dengyihao 已提交
463

dengyihao's avatar
dengyihao 已提交
464 465
  SHashObj*  pPool = pool;
  SConnList* plist = taosHashGet(pPool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
466 467
  if (plist == NULL) {
    SConnList list;
dengyihao's avatar
dengyihao 已提交
468 469
    taosHashPut(pPool, key, strlen(key), (void*)&list, sizeof(list));
    plist = taosHashGet(pPool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
470
    QUEUE_INIT(&plist->conn);
dengyihao's avatar
dengyihao 已提交
471 472 473 474 475 476 477
  }

  if (QUEUE_IS_EMPTY(&plist->conn)) {
    return NULL;
  }
  queue* h = QUEUE_HEAD(&plist->conn);
  QUEUE_REMOVE(h);
dengyihao's avatar
dengyihao 已提交
478
  SCliConn* conn = QUEUE_DATA(h, SCliConn, conn);
U
ubuntu 已提交
479
  conn->status = ConnNormal;
dengyihao's avatar
dengyihao 已提交
480 481
  QUEUE_INIT(&conn->conn);
  return conn;
dengyihao's avatar
dengyihao 已提交
482
}
dengyihao's avatar
dengyihao 已提交
483
static void addConnToPool(void* pool, SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
484 485 486
  SCliThrdObj* thrd = conn->hostThrd;
  CONN_HANDLE_THREAD_QUIT(thrd);

dengyihao's avatar
dengyihao 已提交
487 488 489
  STrans* pTransInst = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
  conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
  transQueueClear(&conn->cliMsgs);
dengyihao's avatar
dengyihao 已提交
490
  transCtxCleanup(&conn->ctx);
dengyihao's avatar
dengyihao 已提交
491
  conn->status = ConnInPool;
dengyihao's avatar
dengyihao 已提交
492

dengyihao's avatar
dengyihao 已提交
493
  char key[128] = {0};
dengyihao's avatar
dengyihao 已提交
494
  CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port);
U
ubuntu 已提交
495
  tTrace("cli conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap);
dengyihao's avatar
dengyihao 已提交
496

dengyihao's avatar
dengyihao 已提交
497
  SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
dengyihao's avatar
dengyihao 已提交
498 499 500
  // list already create before
  assert(plist != NULL);
  QUEUE_PUSH(&plist->conn, &conn->conn);
dengyihao's avatar
dengyihao 已提交
501
  assert(!QUEUE_IS_EMPTY(&plist->conn));
dengyihao's avatar
dengyihao 已提交
502
}
dengyihao's avatar
dengyihao 已提交
503
static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
504
  SCliConn*    conn = handle->data;
dengyihao's avatar
dengyihao 已提交
505
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
506
  transAllocBuffer(pBuf, buf);
dengyihao's avatar
dengyihao 已提交
507
}
U
ubuntu 已提交
508
static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
509
  // impl later
dengyihao's avatar
dengyihao 已提交
510 511 512
  if (handle->data == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
513 514
  SCliConn*    conn = handle->data;
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
515
  if (nread > 0) {
dengyihao's avatar
dengyihao 已提交
516
    pBuf->len += nread;
dengyihao's avatar
dengyihao 已提交
517
    if (transReadComplete(pBuf)) {
U
ubuntu 已提交
518 519
      tTrace("%s cli conn %p read complete", CONN_GET_INST_LABEL(conn), conn);
      cliHandleResp(conn);
dengyihao's avatar
dengyihao 已提交
520
    } else {
U
ubuntu 已提交
521
      tTrace("%s cli conn %p read partial packet, continue to read", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
522
    }
dengyihao's avatar
dengyihao 已提交
523 524
    return;
  }
dengyihao's avatar
dengyihao 已提交
525

526 527
  assert(nread <= 0);
  if (nread == 0) {
dengyihao's avatar
dengyihao 已提交
528 529 530
    // ref http://docs.libuv.org/en/v1.x/stream.html?highlight=uv_read_start#c.uv_read_cb
    // nread might be 0, which does not indicate an error or EOF. This is equivalent to EAGAIN or EWOULDBLOCK under
    // read(2).
dengyihao's avatar
dengyihao 已提交
531
    tTrace("%s cli conn %p read empty", CONN_GET_INST_LABEL(conn), conn);
532 533
    return;
  }
dengyihao's avatar
fix bug  
dengyihao 已提交
534
  if (nread < 0) {
U
ubuntu 已提交
535
    tError("%s cli conn %p read error: %s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(nread));
dengyihao's avatar
dengyihao 已提交
536
    conn->broken = true;
U
ubuntu 已提交
537
    cliHandleExcept(conn);
dengyihao's avatar
dengyihao 已提交
538
  }
dengyihao's avatar
dengyihao 已提交
539
}
dengyihao's avatar
dengyihao 已提交
540

U
ubuntu 已提交
541
static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
wafwerar's avatar
wafwerar 已提交
542
  SCliConn* conn = taosMemoryCalloc(1, sizeof(SCliConn));
dengyihao's avatar
dengyihao 已提交
543
  // read/write stream handle
wafwerar's avatar
wafwerar 已提交
544
  conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
545 546 547 548 549
  uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
  conn->stream->data = conn;

  conn->writeReq.data = conn;
  conn->connReq.data = conn;
dengyihao's avatar
dengyihao 已提交
550 551

  transQueueInit(&conn->cliMsgs, NULL);
dengyihao's avatar
dengyihao 已提交
552 553
  QUEUE_INIT(&conn->conn);
  conn->hostThrd = pThrd;
dengyihao's avatar
dengyihao 已提交
554 555
  conn->status = ConnNormal;
  conn->broken = 0;
dengyihao's avatar
dengyihao 已提交
556 557 558
  transRefCliHandle(conn);
  return conn;
}
U
ubuntu 已提交
559 560
static void cliDestroyConn(SCliConn* conn, bool clear) {
  tTrace("%s cli conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
561

dengyihao's avatar
dengyihao 已提交
562 563
  QUEUE_REMOVE(&conn->conn);
  if (clear) {
U
ubuntu 已提交
564
    uv_close((uv_handle_t*)conn->stream, cliDestroy);
565
  }
dengyihao's avatar
dengyihao 已提交
566
}
U
ubuntu 已提交
567
static void cliDestroy(uv_handle_t* handle) {
dengyihao's avatar
dengyihao 已提交
568
  SCliConn* conn = handle->data;
wafwerar's avatar
wafwerar 已提交
569 570
  taosMemoryFree(conn->ip);
  taosMemoryFree(conn->stream);
dengyihao's avatar
dengyihao 已提交
571 572
  transCtxCleanup(&conn->ctx);
  transQueueDestroy(&conn->cliMsgs);
U
ubuntu 已提交
573
  tTrace("%s cli conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
574
  transDestroyBuffer(&conn->readBuf);
wafwerar's avatar
wafwerar 已提交
575
  taosMemoryFree(conn);
dengyihao's avatar
dengyihao 已提交
576
}
dengyihao's avatar
dengyihao 已提交
577
static bool cliHandleNoResp(SCliConn* conn) {
dengyihao's avatar
dengyihao 已提交
578 579
  bool res = false;
  if (!transQueueEmpty(&conn->cliMsgs)) {
dengyihao's avatar
dengyihao 已提交
580
    SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, 0);
dengyihao's avatar
dengyihao 已提交
581
    if (REQUEST_NO_RESP(&pMsg->msg)) {
dengyihao's avatar
dengyihao 已提交
582 583
      transQueuePop(&conn->cliMsgs);
      // taosArrayRemove(msgs, 0);
dengyihao's avatar
dengyihao 已提交
584 585 586 587 588
      destroyCmsg(pMsg);
      res = true;
    }
    if (res == true) {
      if (cliMaySendCachedMsg(conn) == false) {
dengyihao's avatar
dengyihao 已提交
589 590
        SCliThrdObj* thrd = conn->hostThrd;
        addConnToPool(thrd->pool, conn);
dengyihao's avatar
dengyihao 已提交
591 592 593 594 595
      }
    }
  }
  return res;
}
U
ubuntu 已提交
596
static void cliSendCb(uv_write_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
597
  SCliConn* pConn = req->data;
dengyihao's avatar
dengyihao 已提交
598

dengyihao's avatar
dengyihao 已提交
599
  if (status == 0) {
U
ubuntu 已提交
600
    tTrace("%s cli conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn);
dengyihao's avatar
dengyihao 已提交
601
  } else {
U
ubuntu 已提交
602 603
    tError("%s cli conn %p failed to write: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_err_name(status));
    cliHandleExcept(pConn);
dengyihao's avatar
dengyihao 已提交
604 605
    return;
  }
dengyihao's avatar
dengyihao 已提交
606 607 608 609
  if (cliHandleNoResp(pConn) == true) {
    tTrace("%s cli conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn);
    return;
  }
dengyihao's avatar
dengyihao 已提交
610
  uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb);
dengyihao's avatar
dengyihao 已提交
611
}
dengyihao's avatar
dengyihao 已提交
612

U
ubuntu 已提交
613
void cliSend(SCliConn* pConn) {
dengyihao's avatar
dengyihao 已提交
614 615
  CONN_HANDLE_BROKEN(pConn);

dengyihao's avatar
dengyihao 已提交
616 617
  // assert(taosArrayGetSize(pConn->cliMsgs) > 0);
  assert(!transQueueEmpty(&pConn->cliMsgs));
dengyihao's avatar
dengyihao 已提交
618 619

  SCliMsg* pCliMsg = NULL;
U
ubuntu 已提交
620
  CONN_GET_NEXT_SENDMSG(pConn);
dengyihao's avatar
dengyihao 已提交
621 622
  pCliMsg->sent = 1;

dengyihao's avatar
dengyihao 已提交
623
  STransConnCtx* pCtx = pCliMsg->ctx;
dengyihao's avatar
dengyihao 已提交
624 625

  SCliThrdObj* pThrd = pConn->hostThrd;
dengyihao's avatar
formate  
dengyihao 已提交
626
  STrans*      pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
627

U
ubuntu 已提交
628
  STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg);
dengyihao's avatar
dengyihao 已提交
629 630 631 632
  if (pMsg->pCont == 0) {
    pMsg->pCont = (void*)rpcMallocCont(0);
    pMsg->contLen = 0;
  }
dengyihao's avatar
dengyihao 已提交
633
  int msgLen = transMsgLenFromCont(pMsg->contLen);
dengyihao's avatar
dengyihao 已提交
634

dengyihao's avatar
dengyihao 已提交
635 636
  STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
  pHead->ahandle = pCtx != NULL ? (uint64_t)pCtx->ahandle : 0;
dengyihao's avatar
dengyihao 已提交
637 638
  pHead->noResp = REQUEST_NO_RESP(pMsg) ? 1 : 0;
  pHead->persist = REQUEST_PERSIS_HANDLE(pMsg) ? 1 : 0;
dengyihao's avatar
dengyihao 已提交
639 640
  pHead->msgType = pMsg->msgType;
  pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
dengyihao's avatar
dengyihao 已提交
641
  pHead->release = REQUEST_RELEASE_HANDLE(pCliMsg) ? 1 : 0;
dengyihao's avatar
dengyihao 已提交
642
  memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user));
dengyihao's avatar
dengyihao 已提交
643 644

  uv_buf_t wb = uv_buf_init((char*)pHead, msgLen);
dengyihao's avatar
dengyihao 已提交
645 646
  tDebug("%s cli conn %p %s is send to %s:%d, local info %s:%d", CONN_GET_INST_LABEL(pConn), pConn,
         TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
dengyihao's avatar
dengyihao 已提交
647
         taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port));
dengyihao's avatar
dengyihao 已提交
648

dengyihao's avatar
dengyihao 已提交
649 650 651
  if (pHead->persist == 1) {
    CONN_SET_PERSIST_BY_APP(pConn);
  }
dengyihao's avatar
dengyihao 已提交
652

dengyihao's avatar
dengyihao 已提交
653
  pConn->writeReq.data = pConn;
U
ubuntu 已提交
654
  uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
dengyihao's avatar
dengyihao 已提交
655

U
ubuntu 已提交
656 657
  return;
_RETURN:
dengyihao's avatar
dengyihao 已提交
658
  return;
dengyihao's avatar
dengyihao 已提交
659
}
U
ubuntu 已提交
660 661

void cliConnCb(uv_connect_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
662
  // impl later
dengyihao's avatar
dengyihao 已提交
663 664
  SCliConn* pConn = req->data;
  if (status != 0) {
U
ubuntu 已提交
665 666
    tError("%s cli conn %p failed to connect server: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_strerror(status));
    cliHandleExcept(pConn);
667
    return;
dengyihao's avatar
dengyihao 已提交
668
  }
dengyihao's avatar
dengyihao 已提交
669 670 671
  int addrlen = sizeof(pConn->addr);
  uv_tcp_getpeername((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->addr, &addrlen);

dengyihao's avatar
dengyihao 已提交
672 673
  addrlen = sizeof(pConn->localAddr);
  uv_tcp_getsockname((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->localAddr, &addrlen);
dengyihao's avatar
dengyihao 已提交
674

U
ubuntu 已提交
675
  tTrace("%s cli conn %p connect to server successfully", CONN_GET_INST_LABEL(pConn), pConn);
dengyihao's avatar
dengyihao 已提交
676
  assert(pConn->stream == req->handle);
dengyihao's avatar
dengyihao 已提交
677

U
ubuntu 已提交
678
  cliSend(pConn);
dengyihao's avatar
dengyihao 已提交
679 680
}

U
ubuntu 已提交
681 682
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) {
  tDebug("cli work thread %p start to quit", pThrd);
dengyihao's avatar
dengyihao 已提交
683
  destroyCmsg(pMsg);
dengyihao's avatar
fix bug  
dengyihao 已提交
684
  destroyConnPool(pThrd->pool);
dengyihao's avatar
dengyihao 已提交
685
  uv_timer_stop(&pThrd->timer);
dengyihao's avatar
dengyihao 已提交
686 687
  uv_walk(pThrd->loop, cliWalkCb, NULL);

dengyihao's avatar
dengyihao 已提交
688
  pThrd->quit = true;
dengyihao's avatar
dengyihao 已提交
689

dengyihao's avatar
dengyihao 已提交
690
  // uv_stop(pThrd->loop);
dengyihao's avatar
dengyihao 已提交
691
}
dengyihao's avatar
dengyihao 已提交
692
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
S
Shengliang Guan 已提交
693
  SCliConn* conn = pMsg->msg.info.handle;
dengyihao's avatar
dengyihao 已提交
694
  tDebug("%s cli conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
695

dengyihao's avatar
dengyihao 已提交
696 697
  if (T_REF_VAL_GET(conn) == 2) {
    transUnrefCliHandle(conn);
dengyihao's avatar
dengyihao 已提交
698 699
    if (!transQueuePush(&conn->cliMsgs, pMsg)) {
      return;
dengyihao's avatar
dengyihao 已提交
700 701 702 703 704
    }
    cliSend(conn);
  } else {
    // conn already broken down
    transUnrefCliHandle(conn);
dengyihao's avatar
dengyihao 已提交
705 706
  }
}
dengyihao's avatar
dengyihao 已提交
707 708 709
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd) {
  STransConnCtx* pCtx = pMsg->ctx;

dengyihao's avatar
dengyihao 已提交
710
  pThrd->cvtAddr = pCtx->cvtAddr;
dengyihao's avatar
dengyihao 已提交
711 712
  destroyCmsg(pMsg);
}
U
ubuntu 已提交
713 714

SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) {
dengyihao's avatar
dengyihao 已提交
715
  SCliConn* conn = NULL;
S
Shengliang Guan 已提交
716 717
  if (pMsg->msg.info.handle != NULL) {
    conn = (SCliConn*)(pMsg->msg.info.handle);
dengyihao's avatar
dengyihao 已提交
718
    if (conn != NULL) {
U
ubuntu 已提交
719
      tTrace("%s cli conn %p reused", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
720 721
    }
  } else {
dengyihao's avatar
dengyihao 已提交
722
    STransConnCtx* pCtx = pMsg->ctx;
dengyihao's avatar
dengyihao 已提交
723
    conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet));
dengyihao's avatar
dengyihao 已提交
724 725
    if (conn != NULL) {
      tTrace("%s cli conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn);
dengyihao's avatar
dengyihao 已提交
726 727
    } else {
      tTrace("not found conn in conn pool %p", pThrd->pool);
dengyihao's avatar
dengyihao 已提交
728
    }
dengyihao's avatar
dengyihao 已提交
729
  }
dengyihao's avatar
dengyihao 已提交
730 731
  return conn;
}
dengyihao's avatar
dengyihao 已提交
732 733 734 735 736 737 738 739 740 741 742
void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) {
  if (pCvtAddr->cvt == false) {
    return;
  }
  for (int i = 0; i < pEpSet->numOfEps && pEpSet->numOfEps == 1; i++) {
    if (strncmp(pEpSet->eps[i].fqdn, pCvtAddr->fqdn, TSDB_FQDN_LEN) == 0) {
      memset(pEpSet->eps[i].fqdn, 0, TSDB_FQDN_LEN);
      memcpy(pEpSet->eps[i].fqdn, pCvtAddr->ip, TSDB_FQDN_LEN);
    }
  }
}
U
ubuntu 已提交
743
void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
dengyihao's avatar
dengyihao 已提交
744 745
  uint64_t et = taosGetTimestampUs();
  uint64_t el = et - pMsg->st;
dengyihao's avatar
dengyihao 已提交
746
  // tTrace("%s cli msg tran time cost: %" PRIu64 "us", ((STrans*)pThrd->pTransInst)->label, el);
dengyihao's avatar
dengyihao 已提交
747 748

  STransConnCtx* pCtx = pMsg->ctx;
dengyihao's avatar
formate  
dengyihao 已提交
749
  STrans*        pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
750

dengyihao's avatar
dengyihao 已提交
751
  cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr);
dengyihao's avatar
dengyihao 已提交
752

U
ubuntu 已提交
753
  SCliConn* conn = cliGetConn(pMsg, pThrd);
dengyihao's avatar
dengyihao 已提交
754
  if (conn != NULL) {
dengyihao's avatar
dengyihao 已提交
755
    conn->hThrdIdx = pCtx->hThrdIdx;
dengyihao's avatar
dengyihao 已提交
756

dengyihao's avatar
dengyihao 已提交
757
    transCtxMerge(&conn->ctx, &pCtx->appCtx);
dengyihao's avatar
dengyihao 已提交
758
    transQueuePush(&conn->cliMsgs, pMsg);
U
ubuntu 已提交
759
    cliSend(conn);
dengyihao's avatar
dengyihao 已提交
760
  } else {
U
ubuntu 已提交
761
    conn = cliCreateConn(pThrd);
dengyihao's avatar
dengyihao 已提交
762
    transCtxMerge(&conn->ctx, &pCtx->appCtx);
dengyihao's avatar
dengyihao 已提交
763
    transQueuePush(&conn->cliMsgs, pMsg);
dengyihao's avatar
dengyihao 已提交
764

dengyihao's avatar
dengyihao 已提交
765
    conn->hThrdIdx = pCtx->hThrdIdx;
dengyihao's avatar
dengyihao 已提交
766 767
    conn->ip = strdup(EPSET_GET_INUSE_IP(&pCtx->epSet));
    conn->port = EPSET_GET_INUSE_PORT(&pCtx->epSet);
dengyihao's avatar
dengyihao 已提交
768

dengyihao's avatar
dengyihao 已提交
769 770
    int ret = transSetConnOption((uv_tcp_t*)conn->stream);
    if (ret) {
U
ubuntu 已提交
771
      tError("%s cli conn %p failed to set conn option, errmsg %s", pTransInst->label, conn, uv_err_name(ret));
dengyihao's avatar
dengyihao 已提交
772
    }
dengyihao's avatar
dengyihao 已提交
773 774 775 776 777 778 779
    int fd = taosCreateSocketWithTimeOutOpt(TRANS_CONN_TIMEOUT);
    if (fd == -1) {
      tTrace("%s cli conn %p failed to create socket", pTransInst->label, conn);
      cliHandleExcept(conn);
      return;
    }
    uv_tcp_open((uv_tcp_t*)conn->stream, fd);
780

dengyihao's avatar
dengyihao 已提交
781
    struct sockaddr_in addr;
782 783 784
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = taosGetIpv4FromFqdn(conn->ip);
    addr.sin_port = (uint16_t)htons((uint16_t)conn->port);
dengyihao's avatar
dengyihao 已提交
785 786 787 788 789 790 791 792
    tTrace("%s cli conn %p try to connect to %s:%d", pTransInst->label, conn, conn->ip, conn->port);
    ret = uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, cliConnCb);
    if (ret != 0) {
      tTrace("%s cli conn %p failed to connect to %s:%d, reason: %s", pTransInst->label, conn, conn->ip, conn->port,
             uv_err_name(ret));
      cliHandleExcept(conn);
      return;
    }
dengyihao's avatar
dengyihao 已提交
793 794
  }
}
U
ubuntu 已提交
795
static void cliAsyncCb(uv_async_t* handle) {
dengyihao's avatar
dengyihao 已提交
796 797
  SAsyncItem*  item = handle->data;
  SCliThrdObj* pThrd = item->pThrd;
dengyihao's avatar
dengyihao 已提交
798
  SCliMsg*     pMsg = NULL;
dengyihao's avatar
dengyihao 已提交
799

dengyihao's avatar
dengyihao 已提交
800
  // batch process to avoid to lock/unlock frequently
dengyihao's avatar
formate  
dengyihao 已提交
801
  queue wq;
wafwerar's avatar
wafwerar 已提交
802
  taosThreadMutexLock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
803
  QUEUE_MOVE(&item->qmsg, &wq);
wafwerar's avatar
wafwerar 已提交
804
  taosThreadMutexUnlock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
805

dengyihao's avatar
dengyihao 已提交
806 807 808 809
  int count = 0;
  while (!QUEUE_IS_EMPTY(&wq)) {
    queue* h = QUEUE_HEAD(&wq);
    QUEUE_REMOVE(h);
dengyihao's avatar
dengyihao 已提交
810 811

    SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
dengyihao's avatar
dengyihao 已提交
812 813
    if (pMsg == NULL) {
      continue;
dengyihao's avatar
dengyihao 已提交
814
    }
dengyihao's avatar
dengyihao 已提交
815
    (*cliAsyncHandle[pMsg->type])(pMsg, pThrd);
dengyihao's avatar
dengyihao 已提交
816
    count++;
dengyihao's avatar
dengyihao 已提交
817 818
  }
  if (count >= 2) {
U
ubuntu 已提交
819
    tTrace("cli process batch size: %d", count);
dengyihao's avatar
dengyihao 已提交
820
  }
dengyihao's avatar
dengyihao 已提交
821 822
}

U
ubuntu 已提交
823
static void* cliWorkThread(void* arg) {
dengyihao's avatar
dengyihao 已提交
824
  SCliThrdObj* pThrd = (SCliThrdObj*)arg;
dengyihao's avatar
dengyihao 已提交
825
  pThrd->pid = taosGetSelfPthreadId();
U
ubuntu 已提交
826
  setThreadName("trans-cli-work");
dengyihao's avatar
dengyihao 已提交
827
  uv_run(pThrd->loop, UV_RUN_DEFAULT);
828
  return NULL;
dengyihao's avatar
dengyihao 已提交
829 830
}

U
ubuntu 已提交
831
void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
wafwerar's avatar
wafwerar 已提交
832
  SCliObj* cli = taosMemoryCalloc(1, sizeof(SCliObj));
dengyihao's avatar
dengyihao 已提交
833

dengyihao's avatar
dengyihao 已提交
834
  STrans* pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
835 836
  memcpy(cli->label, label, strlen(label));
  cli->numOfThreads = numOfThreads;
wafwerar's avatar
wafwerar 已提交
837
  cli->pThreadObj = (SCliThrdObj**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrdObj*));
dengyihao's avatar
dengyihao 已提交
838 839

  for (int i = 0; i < cli->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
840
    SCliThrdObj* pThrd = createThrdObj();
dengyihao's avatar
dengyihao 已提交
841
    pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
dengyihao's avatar
dengyihao 已提交
842
    pThrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
843

wafwerar's avatar
wafwerar 已提交
844
    int err = taosThreadCreate(&pThrd->thread, NULL, cliWorkThread, (void*)(pThrd));
dengyihao's avatar
dengyihao 已提交
845
    if (err == 0) {
U
ubuntu 已提交
846
      tDebug("success to create tranport-cli thread %d", i);
dengyihao's avatar
dengyihao 已提交
847
    }
dengyihao's avatar
dengyihao 已提交
848
    cli->pThreadObj[i] = pThrd;
dengyihao's avatar
dengyihao 已提交
849
  }
dengyihao's avatar
dengyihao 已提交
850 851 852 853 854
  int ref = atomic_add_fetch_32(&transSCliInst, 1);
  if (ref == 1) {
    refMgt = transOpenExHandleMgt(50000);
  }

dengyihao's avatar
dengyihao 已提交
855 856
  return cli;
}
dengyihao's avatar
dengyihao 已提交
857

U
ubuntu 已提交
858
static void destroyUserdata(STransMsg* userdata) {
dengyihao's avatar
dengyihao 已提交
859 860 861 862 863 864 865 866 867 868 869 870
  if (userdata->pCont == NULL) {
    return;
  }
  transFreeMsg(userdata->pCont);
  userdata->pCont = NULL;
}
static void destroyCmsg(SCliMsg* pMsg) {
  if (pMsg == NULL) {
    return;
  }
  transDestroyConnCtx(pMsg->ctx);
  destroyUserdata(&pMsg->msg);
wafwerar's avatar
wafwerar 已提交
871
  taosMemoryFree(pMsg);
dengyihao's avatar
dengyihao 已提交
872
}
dengyihao's avatar
dengyihao 已提交
873

dengyihao's avatar
dengyihao 已提交
874
static SCliThrdObj* createThrdObj() {
wafwerar's avatar
wafwerar 已提交
875
  SCliThrdObj* pThrd = (SCliThrdObj*)taosMemoryCalloc(1, sizeof(SCliThrdObj));
U
ubuntu 已提交
876

dengyihao's avatar
dengyihao 已提交
877
  QUEUE_INIT(&pThrd->msg);
wafwerar's avatar
wafwerar 已提交
878
  taosThreadMutexInit(&pThrd->msgMtx, NULL);
dengyihao's avatar
dengyihao 已提交
879

wafwerar's avatar
wafwerar 已提交
880
  pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
881 882
  uv_loop_init(pThrd->loop);

U
ubuntu 已提交
883
  pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 5, pThrd, cliAsyncCb);
dengyihao's avatar
dengyihao 已提交
884 885
  uv_timer_init(pThrd->loop, &pThrd->timer);
  pThrd->timer.data = pThrd;
dengyihao's avatar
dengyihao 已提交
886

dengyihao's avatar
dengyihao 已提交
887
  pThrd->pool = createConnPool(4);
dengyihao's avatar
dengyihao 已提交
888
  transDQCreate(pThrd->loop, &pThrd->delayQueue);
dengyihao's avatar
dengyihao 已提交
889

dengyihao's avatar
dengyihao 已提交
890
  pThrd->quit = false;
dengyihao's avatar
dengyihao 已提交
891 892
  return pThrd;
}
dengyihao's avatar
dengyihao 已提交
893
static void destroyThrdObj(SCliThrdObj* pThrd) {
dengyihao's avatar
dengyihao 已提交
894 895 896
  if (pThrd == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
897

wafwerar's avatar
wafwerar 已提交
898
  taosThreadJoin(pThrd->thread, NULL);
dengyihao's avatar
dengyihao 已提交
899
  CLI_RELEASE_UV(pThrd->loop);
wafwerar's avatar
wafwerar 已提交
900
  taosThreadMutexDestroy(&pThrd->msgMtx);
dengyihao's avatar
dengyihao 已提交
901
  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg);
dengyihao's avatar
dengyihao 已提交
902
  transDestroyAsyncPool(pThrd->asyncPool);
dengyihao's avatar
dengyihao 已提交
903

dengyihao's avatar
dengyihao 已提交
904
  transDQDestroy(pThrd->delayQueue);
wafwerar's avatar
wafwerar 已提交
905 906
  taosMemoryFree(pThrd->loop);
  taosMemoryFree(pThrd);
dengyihao's avatar
dengyihao 已提交
907
}
dengyihao's avatar
dengyihao 已提交
908

dengyihao's avatar
dengyihao 已提交
909
static void transDestroyConnCtx(STransConnCtx* ctx) {
dengyihao's avatar
dengyihao 已提交
910
  //
wafwerar's avatar
wafwerar 已提交
911
  taosMemoryFree(ctx);
dengyihao's avatar
dengyihao 已提交
912
}
dengyihao's avatar
dengyihao 已提交
913

U
ubuntu 已提交
914
void cliSendQuit(SCliThrdObj* thrd) {
dengyihao's avatar
dengyihao 已提交
915
  // cli can stop gracefully
wafwerar's avatar
wafwerar 已提交
916
  SCliMsg* msg = taosMemoryCalloc(1, sizeof(SCliMsg));
dengyihao's avatar
dengyihao 已提交
917
  msg->type = Quit;
dengyihao's avatar
dengyihao 已提交
918
  transSendAsync(thrd->asyncPool, &msg->q);
dengyihao's avatar
dengyihao 已提交
919
}
dengyihao's avatar
dengyihao 已提交
920 921 922 923 924
void cliWalkCb(uv_handle_t* handle, void* arg) {
  if (!uv_is_closing(handle)) {
    uv_close(handle, NULL);
  }
}
dengyihao's avatar
dengyihao 已提交
925

U
ubuntu 已提交
926 927 928 929 930 931 932
int cliRBChoseIdx(STrans* pTransInst) {
  int64_t index = pTransInst->index;
  if (pTransInst->index++ >= pTransInst->numOfThreads) {
    pTransInst->index = 0;
  }
  return index % pTransInst->numOfThreads;
}
dengyihao's avatar
dengyihao 已提交
933 934 935 936 937 938 939 940 941
static void doDelayTask(void* param) {
  STaskArg* arg = param;

  SCliMsg*     pMsg = arg->param1;
  SCliThrdObj* pThrd = arg->param2;
  cliHandleReq(pMsg, pThrd);

  taosMemoryFree(arg);
}
dengyihao's avatar
dengyihao 已提交
942
int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
dengyihao's avatar
dengyihao 已提交
943 944 945
  SCliThrdObj* pThrd = pConn->hostThrd;
  STrans*      pTransInst = pThrd->pTransInst;

dengyihao's avatar
dengyihao 已提交
946 947 948 949 950 951 952 953
  if (pMsg == NULL || pMsg->ctx == NULL) {
    tTrace("%s cli conn %p handle resp", pTransInst->label, pConn);
    pTransInst->cfp(pTransInst->parent, pResp, NULL);
    return 0;
  }

  STransConnCtx* pCtx = pMsg->ctx;
  SEpSet*        pEpSet = &pCtx->epSet;
dengyihao's avatar
dengyihao 已提交
954 955
  transPrintEpSet(pEpSet);

956 957 958
  if (pCtx->retryCount == 0) {
    pCtx->origEpSet = pCtx->epSet;
  }
dengyihao's avatar
dengyihao 已提交
959 960 961
  /*
   * upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL
   */
dengyihao's avatar
dengyihao 已提交
962
  tmsg_t msgType = pCtx->msgType;
963
  if ((pTransInst->retry != NULL && pEpSet->numOfEps > 1 && (pTransInst->retry(pResp->code))) ||
964 965
      (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pResp->code == TSDB_CODE_APP_NOT_READY ||
       pResp->code == TSDB_CODE_NODE_NOT_DEPLOYED || pResp->code == TSDB_CODE_SYN_NOT_LEADER)) {
dengyihao's avatar
dengyihao 已提交
966
    pMsg->sent = 0;
dengyihao's avatar
dengyihao 已提交
967
    pMsg->st = taosGetTimestampUs();
dengyihao's avatar
dengyihao 已提交
968
    pCtx->retryCount += 1;
dengyihao's avatar
dengyihao 已提交
969
    if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
dengyihao's avatar
dengyihao 已提交
970
      if (pCtx->retryCount < pEpSet->numOfEps * 3) {
dengyihao's avatar
dengyihao 已提交
971
        pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
dengyihao's avatar
dengyihao 已提交
972 973 974 975

        STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
        arg->param1 = pMsg;
        arg->param2 = pThrd;
dengyihao's avatar
dengyihao 已提交
976
        transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
dengyihao's avatar
dengyihao 已提交
977 978
        tTrace("use local epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1,
               pEpSet->numOfEps * 3);
dengyihao's avatar
dengyihao 已提交
979
        transUnrefCliHandle(pConn);
dengyihao's avatar
dengyihao 已提交
980 981 982 983
        return -1;
      }
    } else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
      if (pResp->contLen == 0) {
dengyihao's avatar
dengyihao 已提交
984
        pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
dengyihao's avatar
dengyihao 已提交
985 986
        tTrace("use local epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1,
               TRANS_RETRY_COUNT_LIMIT);
dengyihao's avatar
dengyihao 已提交
987
      } else {
988 989 990
        SEpSet epSet = {0};
        tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet);
        pCtx->epSet = epSet;
dengyihao's avatar
dengyihao 已提交
991 992
        tTrace("use remote epset, current in use: %d, retry count:%d, limit: %d", pEpSet->inUse, pCtx->retryCount + 1,
               TRANS_RETRY_COUNT_LIMIT);
dengyihao's avatar
dengyihao 已提交
993
      }
dengyihao's avatar
dengyihao 已提交
994
      addConnToPool(pThrd->pool, pConn);
dengyihao's avatar
dengyihao 已提交
995

dengyihao's avatar
dengyihao 已提交
996 997 998
      STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
      arg->param1 = pMsg;
      arg->param2 = pThrd;
dengyihao's avatar
dengyihao 已提交
999
      transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
dengyihao's avatar
dengyihao 已提交
1000
      return -1;
dengyihao's avatar
dengyihao 已提交
1001 1002 1003 1004
    }
  }

  if (pCtx->pSem != NULL) {
1005 1006 1007 1008 1009 1010
    tTrace("%s cli conn %p(sync) handle resp", pTransInst->label, pConn);
    if (pCtx->pRsp == NULL) {
      tTrace("%s cli conn %p(sync) failed to resp, ignore", pTransInst->label, pConn);
    } else {
      memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp));
    }
dengyihao's avatar
dengyihao 已提交
1011
    tsem_post(pCtx->pSem);
1012
    pCtx->pRsp = NULL;
dengyihao's avatar
dengyihao 已提交
1013
  } else {
dengyihao's avatar
dengyihao 已提交
1014
    tTrace("%s cli conn %p handle resp", pTransInst->label, pConn);
1015
    if (pResp->code != 0 || pCtx->retryCount == 0 || transEpSetIsEqual(&pCtx->epSet, &pCtx->origEpSet)) {
dengyihao's avatar
dengyihao 已提交
1016 1017 1018 1019
      pTransInst->cfp(pTransInst->parent, pResp, NULL);
    } else {
      pTransInst->cfp(pTransInst->parent, pResp, pEpSet);
    }
dengyihao's avatar
dengyihao 已提交
1020
  }
dengyihao's avatar
dengyihao 已提交
1021
  return 0;
dengyihao's avatar
dengyihao 已提交
1022
}
U
ubuntu 已提交
1023 1024

void transCloseClient(void* arg) {
U
ubuntu 已提交
1025
  SCliObj* cli = arg;
dengyihao's avatar
dengyihao 已提交
1026
  for (int i = 0; i < cli->numOfThreads; i++) {
U
ubuntu 已提交
1027
    cliSendQuit(cli->pThreadObj[i]);
dengyihao's avatar
dengyihao 已提交
1028
    destroyThrdObj(cli->pThreadObj[i]);
dengyihao's avatar
dengyihao 已提交
1029
  }
wafwerar's avatar
wafwerar 已提交
1030 1031
  taosMemoryFree(cli->pThreadObj);
  taosMemoryFree(cli);
dengyihao's avatar
dengyihao 已提交
1032 1033 1034 1035
  int ref = atomic_sub_fetch_32(&transSCliInst, 1);
  if (ref == 0) {
    transCloseExHandleMgt(refMgt);
  }
dengyihao's avatar
dengyihao 已提交
1036
}
dengyihao's avatar
dengyihao 已提交
1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048
void transRefCliHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
  int ref = T_REF_INC((SCliConn*)handle);
  UNUSED(ref);
}
void transUnrefCliHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
  int ref = T_REF_DEC((SCliConn*)handle);
dengyihao's avatar
dengyihao 已提交
1049
  tDebug("%s cli conn %p ref %d", CONN_GET_INST_LABEL((SCliConn*)handle), handle, ref);
dengyihao's avatar
dengyihao 已提交
1050
  if (ref == 0) {
U
ubuntu 已提交
1051
    cliDestroyConn((SCliConn*)handle, true);
dengyihao's avatar
dengyihao 已提交
1052 1053
  }
}
dengyihao's avatar
dengyihao 已提交
1054 1055 1056 1057 1058 1059
void transReleaseCliHandle(void* handle) {
  SCliThrdObj* thrd = CONN_GET_HOST_THREAD(handle);
  if (thrd == NULL) {
    return;
  }

S
Shengliang Guan 已提交
1060
  STransMsg tmsg = {.info.handle = handle};
wafwerar's avatar
wafwerar 已提交
1061
  SCliMsg*  cmsg = taosMemoryCalloc(1, sizeof(SCliMsg));
dengyihao's avatar
dengyihao 已提交
1062
  cmsg->msg = tmsg;
dengyihao's avatar
dengyihao 已提交
1063
  cmsg->type = Release;
dengyihao's avatar
dengyihao 已提交
1064 1065 1066

  transSendAsync(thrd->asyncPool, &cmsg->q);
}
dengyihao's avatar
dengyihao 已提交
1067

dengyihao's avatar
dengyihao 已提交
1068
void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
U
ubuntu 已提交
1069
  STrans* pTransInst = (STrans*)shandle;
S
Shengliang Guan 已提交
1070
  int     index = CONN_HOST_THREAD_INDEX((SCliConn*)pReq->info.handle);
dengyihao's avatar
dengyihao 已提交
1071
  if (index == -1) {
U
ubuntu 已提交
1072
    index = cliRBChoseIdx(pTransInst);
dengyihao's avatar
dengyihao 已提交
1073
  }
dengyihao's avatar
dengyihao 已提交
1074

wafwerar's avatar
wafwerar 已提交
1075
  STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
dengyihao's avatar
dengyihao 已提交
1076
  pCtx->epSet = *pEpSet;
S
Shengliang Guan 已提交
1077
  pCtx->ahandle = pReq->info.ahandle;
dengyihao's avatar
dengyihao 已提交
1078
  pCtx->msgType = pReq->msgType;
dengyihao's avatar
dengyihao 已提交
1079
  pCtx->hThrdIdx = index;
dengyihao's avatar
dengyihao 已提交
1080 1081 1082 1083

  if (ctx != NULL) {
    pCtx->appCtx = *ctx;
  }
dengyihao's avatar
dengyihao 已提交
1084
  assert(pTransInst->connType == TAOS_CONN_CLIENT);
dengyihao's avatar
dengyihao 已提交
1085

wafwerar's avatar
wafwerar 已提交
1086
  SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
dengyihao's avatar
dengyihao 已提交
1087
  cliMsg->ctx = pCtx;
dengyihao's avatar
dengyihao 已提交
1088
  cliMsg->msg = *pReq;
dengyihao's avatar
dengyihao 已提交
1089
  cliMsg->st = taosGetTimestampUs();
dengyihao's avatar
dengyihao 已提交
1090
  cliMsg->type = Normal;
dengyihao's avatar
dengyihao 已提交
1091

U
ubuntu 已提交
1092
  SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[index];
dengyihao's avatar
dengyihao 已提交
1093

dengyihao's avatar
dengyihao 已提交
1094
  tDebug("send request at thread:%d, threadID: %08" PRId64 ",  msg: %p, dst: %s:%d, app:%p", index, thrd->pid, pReq,
S
Shengliang Guan 已提交
1095
         EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
L
Liu Jicong 已提交
1096
  ASSERT(transSendAsync(thrd->asyncPool, &(cliMsg->q)) == 0);
dengyihao's avatar
dengyihao 已提交
1097
}
dengyihao's avatar
dengyihao 已提交
1098

dengyihao's avatar
dengyihao 已提交
1099
void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) {
U
ubuntu 已提交
1100
  STrans* pTransInst = (STrans*)shandle;
S
Shengliang Guan 已提交
1101
  int     index = CONN_HOST_THREAD_INDEX(pReq->info.handle);
dengyihao's avatar
dengyihao 已提交
1102
  if (index == -1) {
U
ubuntu 已提交
1103
    index = cliRBChoseIdx(pTransInst);
dengyihao's avatar
dengyihao 已提交
1104
  }
dengyihao's avatar
dengyihao 已提交
1105 1106
  tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t));
  tsem_init(sem, 0, 0);
dengyihao's avatar
dengyihao 已提交
1107

wafwerar's avatar
wafwerar 已提交
1108
  STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
dengyihao's avatar
dengyihao 已提交
1109
  pCtx->epSet = *pEpSet;
S
Shengliang Guan 已提交
1110
  pCtx->ahandle = pReq->info.ahandle;
dengyihao's avatar
dengyihao 已提交
1111
  pCtx->msgType = pReq->msgType;
dengyihao's avatar
dengyihao 已提交
1112
  pCtx->hThrdIdx = index;
dengyihao's avatar
dengyihao 已提交
1113
  pCtx->pSem = sem;
dengyihao's avatar
dengyihao 已提交
1114 1115
  pCtx->pRsp = pRsp;

wafwerar's avatar
wafwerar 已提交
1116
  SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
dengyihao's avatar
dengyihao 已提交
1117 1118 1119
  cliMsg->ctx = pCtx;
  cliMsg->msg = *pReq;
  cliMsg->st = taosGetTimestampUs();
dengyihao's avatar
dengyihao 已提交
1120
  cliMsg->type = Normal;
dengyihao's avatar
dengyihao 已提交
1121

U
ubuntu 已提交
1122
  SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[index];
dengyihao's avatar
dengyihao 已提交
1123
  tDebug("send request at thread:%d, threadID:%08" PRId64 ", msg: %p, dst: %s:%d, app:%p", index, thrd->pid, pReq,
S
Shengliang Guan 已提交
1124
         EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
dengyihao's avatar
dengyihao 已提交
1125

dengyihao's avatar
dengyihao 已提交
1126
  transSendAsync(thrd->asyncPool, &(cliMsg->q));
dengyihao's avatar
dengyihao 已提交
1127 1128 1129
  tsem_wait(sem);
  tsem_destroy(sem);
  taosMemoryFree(sem);
dengyihao's avatar
dengyihao 已提交
1130
}
U
ubuntu 已提交
1131

dengyihao's avatar
dengyihao 已提交
1132 1133 1134 1135
/*
 *
 **/
void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) {
dengyihao's avatar
dengyihao 已提交
1136
  STrans* pTransInst = ahandle;
dengyihao's avatar
dengyihao 已提交
1137 1138 1139 1140 1141 1142 1143

  SCvtAddr cvtAddr = {0};
  if (ip != NULL && fqdn != NULL) {
    memcpy(cvtAddr.ip, ip, strlen(ip));
    memcpy(cvtAddr.fqdn, fqdn, strlen(fqdn));
    cvtAddr.cvt = true;
  }
dengyihao's avatar
dengyihao 已提交
1144 1145 1146
  for (int i = 0; i < pTransInst->numOfThreads; i++) {
    STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
    pCtx->hThrdIdx = i;
dengyihao's avatar
dengyihao 已提交
1147
    pCtx->cvtAddr = cvtAddr;
dengyihao's avatar
dengyihao 已提交
1148 1149 1150 1151 1152 1153

    SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
    cliMsg->ctx = pCtx;
    cliMsg->type = Update;

    SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
dengyihao's avatar
dengyihao 已提交
1154
    tDebug("update epset at thread:%d, threadID:%08" PRId64 "", i, thrd->pid);
dengyihao's avatar
dengyihao 已提交
1155 1156 1157 1158

    transSendAsync(thrd->asyncPool, &(cliMsg->q));
  }
}
dengyihao's avatar
dengyihao 已提交
1159
#endif