transSvr.c 39.7 KB
Newer Older
dengyihao's avatar
dengyihao 已提交
1 2
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
dengyihao's avatar
dengyihao 已提交
3
 * * This program is free software: you can use, redistribute, and/or modify
dengyihao's avatar
dengyihao 已提交
4 5 6 7 8 9 10 11 12 13 14
 * 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/>.
 */

15
#include "transComm.h"
dengyihao's avatar
dengyihao 已提交
16

dengyihao's avatar
dengyihao 已提交
17 18
static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;

dengyihao's avatar
dengyihao 已提交
19
static char* notify = "a";
dengyihao's avatar
dengyihao 已提交
20

dengyihao's avatar
dengyihao 已提交
21 22 23 24
typedef struct {
  int       notifyCount;  //
  int       init;         // init or not
  STransMsg msg;
dengyihao's avatar
dengyihao 已提交
25
} SSvrRegArg;
dengyihao's avatar
dengyihao 已提交
26

dengyihao's avatar
dengyihao 已提交
27
typedef struct SSvrConn {
dengyihao's avatar
dengyihao 已提交
28
  T_REF_DECLARE()
dengyihao's avatar
dengyihao 已提交
29
  uv_tcp_t*  pTcp;
dengyihao's avatar
dengyihao 已提交
30
  queue      wreqQueue;
dengyihao's avatar
dengyihao 已提交
31
  uv_timer_t pTimer;
dengyihao's avatar
dengyihao 已提交
32 33

  queue       queue;
dengyihao's avatar
dengyihao 已提交
34
  SConnBuffer readBuf;  // read buf,
dengyihao's avatar
dengyihao 已提交
35 36 37
  int         inType;
  void*       pTransInst;  // rpc init
  void*       ahandle;     //
dengyihao's avatar
dengyihao 已提交
38
  void*       hostThrd;
dengyihao's avatar
dengyihao 已提交
39
  STransQueue srvMsgs;
dengyihao's avatar
dengyihao 已提交
40

dengyihao's avatar
dengyihao 已提交
41
  SSvrRegArg regArg;
dengyihao's avatar
dengyihao 已提交
42
  bool       broken;  // conn broken;
dengyihao's avatar
dengyihao 已提交
43

dengyihao's avatar
dengyihao 已提交
44 45 46 47 48 49 50
  ConnStatus status;

  uint32_t clientIp;
  uint16_t port;

  char src[32];
  char dst[32];
U
ubuntu 已提交
51

dengyihao's avatar
dengyihao 已提交
52 53 54 55 56 57
  int64_t refId;
  int     spi;
  char    info[64];
  char    user[TSDB_UNI_LEN];  // user ID for the link
  char    secret[TSDB_PASSWORD_LEN];
  char    ckey[TSDB_PASSWORD_LEN];  // ciphering key
dengyihao's avatar
dengyihao 已提交
58
} SSvrConn;
dengyihao's avatar
dengyihao 已提交
59

dengyihao's avatar
dengyihao 已提交
60 61
typedef struct SSvrMsg {
  SSvrConn*     pConn;
dengyihao's avatar
dengyihao 已提交
62 63 64
  STransMsg     msg;
  queue         q;
  STransMsgType type;
dengyihao's avatar
dengyihao 已提交
65
} SSvrMsg;
dengyihao's avatar
dengyihao 已提交
66

dengyihao's avatar
dengyihao 已提交
67
typedef struct SWorkThrd {
dengyihao's avatar
dengyihao 已提交
68
  TdThread      thread;
wafwerar's avatar
wafwerar 已提交
69
  uv_connect_t  connect_req;
dengyihao's avatar
dengyihao 已提交
70 71 72 73
  uv_pipe_t*    pipe;
  uv_os_fd_t    fd;
  uv_loop_t*    loop;
  SAsyncPool*   asyncPool;
dengyihao's avatar
dengyihao 已提交
74
  uv_prepare_t* prepare;
dengyihao's avatar
dengyihao 已提交
75
  queue         msg;
dengyihao's avatar
dengyihao 已提交
76 77 78 79

  queue conn;
  void* pTransInst;
  bool  quit;
dengyihao's avatar
dengyihao 已提交
80
} SWorkThrd;
dengyihao's avatar
dengyihao 已提交
81 82

typedef struct SServerObj {
dengyihao's avatar
dengyihao 已提交
83
  TdThread   thread;
dengyihao's avatar
dengyihao 已提交
84 85 86 87
  uv_tcp_t   server;
  uv_loop_t* loop;

  // work thread info
dengyihao's avatar
dengyihao 已提交
88 89 90 91
  int         workerIdx;
  int         numOfThreads;
  int         numOfWorkerReady;
  SWorkThrd** pThreadObj;
dengyihao's avatar
dengyihao 已提交
92

wafwerar's avatar
wafwerar 已提交
93
  uv_pipe_t   pipeListen;
dengyihao's avatar
dengyihao 已提交
94 95 96 97
  uv_pipe_t** pipe;
  uint32_t    ip;
  uint32_t    port;
  uv_async_t* pAcceptAsync;  // just to quit from from accept thread
dengyihao's avatar
dengyihao 已提交
98 99

  bool inited;
dengyihao's avatar
dengyihao 已提交
100 101 102
} SServerObj;

static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
103
static void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
104
static void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
105
static void uvOnTimeoutCb(uv_timer_t* handle);
dengyihao's avatar
dengyihao 已提交
106
static void uvOnSendCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
107
static void uvOnPipeWriteCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
108 109 110
static void uvOnAcceptCb(uv_stream_t* stream, int status);
static void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf);
static void uvWorkerAsyncCb(uv_async_t* handle);
dengyihao's avatar
dengyihao 已提交
111
static void uvAcceptAsyncCb(uv_async_t* handle);
dengyihao's avatar
dengyihao 已提交
112
static void uvShutDownCb(uv_shutdown_t* req, int status);
dengyihao's avatar
dengyihao 已提交
113
static void uvPrepareCb(uv_prepare_t* handle);
dengyihao's avatar
dengyihao 已提交
114

dengyihao's avatar
dengyihao 已提交
115 116
static bool uvRecvReleaseReq(SSvrConn* conn, STransMsgHead* pHead);

dengyihao's avatar
dengyihao 已提交
117 118 119 120 121 122
/*
 * time-consuming task throwed into BG work thread
 */
static void uvWorkDoTask(uv_work_t* req);
static void uvWorkAfterTask(uv_work_t* req, int status);

dengyihao's avatar
dengyihao 已提交
123 124
static void uvWalkCb(uv_handle_t* handle, void* arg);
static void uvFreeCb(uv_handle_t* handle);
dengyihao's avatar
dengyihao 已提交
125

dengyihao's avatar
dengyihao 已提交
126 127
static FORCE_INLINE void uvStartSendRespImpl(SSvrMsg* smsg);

dengyihao's avatar
dengyihao 已提交
128
static int  uvPrepareSendData(SSvrMsg* msg, uv_buf_t* wb);
dengyihao's avatar
dengyihao 已提交
129
static void uvStartSendResp(SSvrMsg* msg);
dengyihao's avatar
dengyihao 已提交
130

dengyihao's avatar
dengyihao 已提交
131
static void uvNotifyLinkBrokenToApp(SSvrConn* conn);
dengyihao's avatar
dengyihao 已提交
132

dengyihao's avatar
dengyihao 已提交
133 134 135 136
static FORCE_INLINE void      destroySmsg(SSvrMsg* smsg);
static FORCE_INLINE SSvrConn* createConn(void* hThrd);
static FORCE_INLINE void      destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/);
static FORCE_INLINE void      destroyConnRegArg(SSvrConn* conn);
dengyihao's avatar
dengyihao 已提交
137

dengyihao's avatar
dengyihao 已提交
138
static int reallocConnRef(SSvrConn* conn);
dengyihao's avatar
dengyihao 已提交
139

dengyihao's avatar
dengyihao 已提交
140 141 142 143 144 145
static void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd);
static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrd* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
                                                                    uvHandleRegister, NULL};
dengyihao's avatar
dengyihao 已提交
146

147
static void uvDestroyConn(uv_handle_t* handle);
dengyihao's avatar
dengyihao 已提交
148

dengyihao's avatar
dengyihao 已提交
149
// server and worker thread
150 151
static void* transWorkerThread(void* arg);
static void* transAcceptThread(void* arg);
dengyihao's avatar
dengyihao 已提交
152

dengyihao's avatar
dengyihao 已提交
153
// add handle loop
dengyihao's avatar
dengyihao 已提交
154
static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
dengyihao's avatar
dengyihao 已提交
155 156
static bool addHandleToAcceptloop(void* arg);

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

dengyihao's avatar
dengyihao 已提交
164 165 166 167
#define ASYNC_ERR_JRET(thrd)                            \
  do {                                                  \
    if (thrd->quit) {                                   \
      tTrace("worker thread already quit, ignore msg"); \
dengyihao's avatar
dengyihao 已提交
168
      goto _return1;                                    \
dengyihao's avatar
dengyihao 已提交
169 170 171
    }                                                   \
  } while (0)

dengyihao's avatar
dengyihao 已提交
172
void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
173
  SSvrConn*    conn = handle->data;
dengyihao's avatar
dengyihao 已提交
174 175
  SConnBuffer* pBuf = &conn->readBuf;
  transAllocBuffer(pBuf, buf);
dengyihao's avatar
dengyihao 已提交
176 177 178 179
}

// refers specifically to query or insert timeout
static void uvHandleActivityTimeout(uv_timer_t* handle) {
dengyihao's avatar
dengyihao 已提交
180
  SSvrConn* conn = handle->data;
dengyihao's avatar
dengyihao 已提交
181
  tDebug("%p timeout since no activity", conn);
dengyihao's avatar
dengyihao 已提交
182 183
}

dengyihao's avatar
dengyihao 已提交
184
static bool uvHandleReq(SSvrConn* pConn) {
dengyihao's avatar
dengyihao 已提交
185
  STrans* pTransInst = pConn->pTransInst;
dengyihao's avatar
opt rpc  
dengyihao 已提交
186

dengyihao's avatar
dengyihao 已提交
187 188 189
  STransMsgHead* pHead = NULL;

  int msgLen = transDumpFromBuffer(&pConn->readBuf, (char**)&pHead);
dengyihao's avatar
dengyihao 已提交
190
  if (msgLen <= 0) {
dengyihao's avatar
dengyihao 已提交
191 192
    tError("%s conn %p read invalid packet", transLabel(pTransInst), pConn);
    return false;
dengyihao's avatar
dengyihao 已提交
193
  }
dengyihao's avatar
dengyihao 已提交
194

dengyihao's avatar
dengyihao 已提交
195
  if (transDecompressMsg((char**)&pHead, msgLen) < 0) {
dengyihao's avatar
dengyihao 已提交
196
    tError("%s conn %p recv invalid packet, failed to decompress", transLabel(pTransInst), pConn);
dengyihao's avatar
dengyihao 已提交
197 198 199
    return false;
  }

dengyihao's avatar
dengyihao 已提交
200
  pHead->code = htonl(pHead->code);
dengyihao's avatar
dengyihao 已提交
201
  pHead->msgLen = htonl(pHead->msgLen);
dengyihao's avatar
dengyihao 已提交
202

dengyihao's avatar
dengyihao 已提交
203
  memcpy(pConn->user, pHead->user, strlen(pHead->user));
dengyihao's avatar
dengyihao 已提交
204

dengyihao's avatar
dengyihao 已提交
205 206 207 208
  if (uvRecvReleaseReq(pConn, pHead)) {
    return true;
  }

dengyihao's avatar
dengyihao 已提交
209 210 211 212 213
  // TODO(dengyihao): time-consuming task throwed into BG Thread
  //  uv_work_t* wreq = taosMemoryMalloc(sizeof(uv_work_t));
  //  wreq->data = pConn;
  //  uv_read_stop((uv_stream_t*)pConn->pTcp);
  //  transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
214
  //  uv_queue_work(((SWorkThrd*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask);
dengyihao's avatar
dengyihao 已提交
215

dengyihao's avatar
dengyihao 已提交
216
  STransMsg transMsg;
dengyihao's avatar
dengyihao 已提交
217
  memset(&transMsg, 0, sizeof(transMsg));
dengyihao's avatar
dengyihao 已提交
218 219 220 221
  transMsg.contLen = transContLenFromMsg(pHead->msgLen);
  transMsg.pCont = pHead->content;
  transMsg.msgType = pHead->msgType;
  transMsg.code = pHead->code;
dengyihao's avatar
dengyihao 已提交
222

dengyihao's avatar
dengyihao 已提交
223
  pConn->inType = pHead->msgType;
dengyihao's avatar
dengyihao 已提交
224 225 226 227
  if (pConn->status == ConnNormal) {
    if (pHead->persist == 1) {
      pConn->status = ConnAcquire;
      transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
228
      tDebug("conn %p acquired by server app", pConn);
dengyihao's avatar
dengyihao 已提交
229 230
    }
  }
dengyihao's avatar
dengyihao 已提交
231
  STraceId* trace = &pHead->traceId;
dengyihao's avatar
dengyihao 已提交
232 233 234 235

  int64_t        cost = taosGetTimestampUs() - taosNtoh64(pHead->timestamp);
  static int64_t EXCEPTION_LIMIT_US = 100 * 1000;

dengyihao's avatar
dengyihao 已提交
236
  if (pConn->status == ConnNormal && pHead->noResp == 0) {
dengyihao's avatar
dengyihao 已提交
237
    transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
238 239 240 241 242 243 244
    if (cost >= EXCEPTION_LIMIT_US) {
      tGWarn("%s conn %p %s received from %s, local info:%s, len:%d, cost:%dus, recv exception", transLabel(pTransInst),
             pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, (int)cost);
    } else {
      tGDebug("%s conn %p %s received from %s, local info:%s, len:%d, cost:%dus", transLabel(pTransInst), pConn,
              TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, (int)cost);
    }
dengyihao's avatar
dengyihao 已提交
245
  } else {
dengyihao's avatar
dengyihao 已提交
246
    if (cost >= EXCEPTION_LIMIT_US) {
dengyihao's avatar
dengyihao 已提交
247
      tGWarn("%s conn %p %s received from %s, local info:%s, len:%d, noResp:%d, code:%d, cost:%dus, recv exception",
dengyihao's avatar
dengyihao 已提交
248 249 250
             transLabel(pTransInst), pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, pHead->noResp,
             transMsg.code, (int)(cost));
    } else {
dengyihao's avatar
dengyihao 已提交
251
      tGDebug("%s conn %p %s received from %s, local info:%s, len:%d, noResp:%d, code:%d, cost:%dus",
dengyihao's avatar
dengyihao 已提交
252 253
              transLabel(pTransInst), pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, pHead->noResp,
              transMsg.code, (int)(cost));
dengyihao's avatar
dengyihao 已提交
254
    }
dengyihao's avatar
dengyihao 已提交
255 256
  }

dengyihao's avatar
dengyihao 已提交
257
  // pHead->noResp = 1,
dengyihao's avatar
dengyihao 已提交
258 259 260
  // 1. server application should not send resp on handle
  // 2. once send out data, cli conn released to conn pool immediately
  // 3. not mixed with persist
dengyihao's avatar
dengyihao 已提交
261
  transMsg.info.ahandle = (void*)pHead->ahandle;
dengyihao's avatar
dengyihao 已提交
262
  transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId);
S
Shengliang Guan 已提交
263
  transMsg.info.refId = pConn->refId;
dengyihao's avatar
dengyihao 已提交
264 265
  transMsg.info.traceId = pHead->traceId;

dengyihao's avatar
dengyihao 已提交
266 267
  tGTrace("%s handle %p conn:%p translated to app, refId:%" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn,
          pConn->refId);
dengyihao's avatar
dengyihao 已提交
268 269 270 271
  ASSERTS(transMsg.info.handle != NULL, "trans-svr failed to alloc handle to msg");
  if (transMsg.info.handle == NULL) {
    return false;
  }
dengyihao's avatar
dengyihao 已提交
272

dengyihao's avatar
dengyihao 已提交
273
  if (pHead->noResp == 1) {
S
Shengliang Guan 已提交
274
    transMsg.info.refId = -1;
dengyihao's avatar
dengyihao 已提交
275
  }
dengyihao's avatar
dengyihao 已提交
276 277

  // set up conn info
278
  SRpcConnInfo* pConnInfo = &(transMsg.info.conn);
dengyihao's avatar
dengyihao 已提交
279 280
  pConnInfo->clientIp = pConn->clientIp;
  pConnInfo->clientPort = pConn->port;
dengyihao's avatar
dengyihao 已提交
281 282
  tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user));

dengyihao's avatar
dengyihao 已提交
283
  transReleaseExHandle(transGetRefMgt(), pConn->refId);
dengyihao's avatar
dengyihao 已提交
284

dengyihao's avatar
dengyihao 已提交
285
  (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
dengyihao's avatar
dengyihao 已提交
286
  return true;
dengyihao's avatar
dengyihao 已提交
287 288
}

dengyihao's avatar
dengyihao 已提交
289
void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
290 291 292
  SSvrConn* conn = cli->data;
  STrans*   pTransInst = conn->pTransInst;

dengyihao's avatar
dengyihao 已提交
293
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
294 295
  if (nread > 0) {
    pBuf->len += nread;
dengyihao's avatar
dengyihao 已提交
296 297
    if (pBuf->len <= TRANS_PACKET_LIMIT) {
      while (transReadComplete(pBuf)) {
dengyihao's avatar
dengyihao 已提交
298
        if (true == pBuf->invalid || false == uvHandleReq(conn)) {
dengyihao's avatar
dengyihao 已提交
299 300
          tError("%s conn %p read invalid packet, received from %s, local info:%s", transLabel(pTransInst), conn,
                 conn->dst, conn->src);
dengyihao's avatar
dengyihao 已提交
301
          destroyConn(conn, true);
dengyihao's avatar
dengyihao 已提交
302
          return;
dengyihao's avatar
dengyihao 已提交
303
        }
dengyihao's avatar
dengyihao 已提交
304 305 306
      }
      return;
    } else {
dengyihao's avatar
dengyihao 已提交
307
      tError("%s conn %p read invalid packet, exceed limit, received from %s, local info:%s", transLabel(pTransInst),
dengyihao's avatar
dengyihao 已提交
308
             conn, conn->dst, conn->src);
dengyihao's avatar
dengyihao 已提交
309 310
      destroyConn(conn, true);
      return;
dengyihao's avatar
dengyihao 已提交
311 312
    }
  }
313 314 315
  if (nread == 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
316

317
  tWarn("%s conn %p read error:%s", transLabel(pTransInst), conn, uv_err_name(nread));
U
ubuntu 已提交
318
  if (nread < 0) {
dengyihao's avatar
dengyihao 已提交
319
    conn->broken = true;
dengyihao's avatar
dengyihao 已提交
320 321
    if (conn->status == ConnAcquire) {
      if (conn->regArg.init) {
dengyihao's avatar
dengyihao 已提交
322
        tTrace("%s conn %p broken, notify server app", transLabel(pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
323 324 325 326 327
        STrans* pTransInst = conn->pTransInst;
        (*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
        memset(&conn->regArg, 0, sizeof(conn->regArg));
      }
    }
dengyihao's avatar
dengyihao 已提交
328
    destroyConn(conn, true);
dengyihao's avatar
dengyihao 已提交
329 330 331 332
  }
}
void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
  buf->len = 2;
wafwerar's avatar
wafwerar 已提交
333
  buf->base = taosMemoryCalloc(1, sizeof(char) * buf->len);
dengyihao's avatar
dengyihao 已提交
334 335 336 337
}

void uvOnTimeoutCb(uv_timer_t* handle) {
  // opt
dengyihao's avatar
dengyihao 已提交
338
  SSvrConn* pConn = handle->data;
dengyihao's avatar
dengyihao 已提交
339
  tError("conn %p time out", pConn);
dengyihao's avatar
dengyihao 已提交
340 341
}

dengyihao's avatar
dengyihao 已提交
342
void uvOnSendCb(uv_write_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
343
  SSvrConn* conn = transReqQueueRemove(req);
dengyihao's avatar
dengyihao 已提交
344 345
  if (conn == NULL) return;

dengyihao's avatar
dengyihao 已提交
346
  if (status == 0) {
dengyihao's avatar
dengyihao 已提交
347
    tTrace("conn %p data already was written on stream", conn);
dengyihao's avatar
dengyihao 已提交
348
    if (!transQueueEmpty(&conn->srvMsgs)) {
dengyihao's avatar
dengyihao 已提交
349 350 351 352
      SSvrMsg*  msg = transQueuePop(&conn->srvMsgs);
      STraceId* trace = &msg->msg.info.traceId;
      tGDebug("conn %p write data out", conn);

dengyihao's avatar
add UT  
dengyihao 已提交
353
      destroySmsg(msg);
dengyihao's avatar
dengyihao 已提交
354
      // send cached data
dengyihao's avatar
dengyihao 已提交
355
      if (!transQueueEmpty(&conn->srvMsgs)) {
dengyihao's avatar
dengyihao 已提交
356
        msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
dengyihao's avatar
dengyihao 已提交
357 358 359 360 361 362 363 364 365
        if (msg->type == Register && conn->status == ConnAcquire) {
          conn->regArg.notifyCount = 0;
          conn->regArg.init = 1;
          conn->regArg.msg = msg->msg;
          if (conn->broken) {
            STrans* pTransInst = conn->pTransInst;
            (pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
            memset(&conn->regArg, 0, sizeof(conn->regArg));
          }
dengyihao's avatar
dengyihao 已提交
366
          transQueuePop(&conn->srvMsgs);
wafwerar's avatar
wafwerar 已提交
367
          taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
368

dengyihao's avatar
dengyihao 已提交
369
          msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
dengyihao's avatar
dengyihao 已提交
370
          if (msg != NULL) {
dengyihao's avatar
dengyihao 已提交
371
            uvStartSendRespImpl(msg);
dengyihao's avatar
dengyihao 已提交
372
          }
dengyihao's avatar
dengyihao 已提交
373
        } else {
dengyihao's avatar
dengyihao 已提交
374
          uvStartSendRespImpl(msg);
dengyihao's avatar
dengyihao 已提交
375
        }
dengyihao's avatar
add UT  
dengyihao 已提交
376
      }
dengyihao's avatar
dengyihao 已提交
377
    }
dengyihao's avatar
dengyihao 已提交
378
    transUnrefSrvHandle(conn);
dengyihao's avatar
dengyihao 已提交
379
  } else {
dengyihao's avatar
dengyihao 已提交
380 381 382 383 384
    if (!uv_is_closing((uv_handle_t*)(conn->pTcp))) {
      tError("conn %p failed to write data, %s", conn, uv_err_name(status));
      conn->broken = true;
      transUnrefSrvHandle(conn);
    }
dengyihao's avatar
dengyihao 已提交
385 386
  }
}
dengyihao's avatar
dengyihao 已提交
387 388
static void uvOnPipeWriteCb(uv_write_t* req, int status) {
  if (status == 0) {
dengyihao's avatar
dengyihao 已提交
389
    tTrace("success to dispatch conn to work thread");
dengyihao's avatar
dengyihao 已提交
390 391 392
  } else {
    tError("fail to dispatch conn to work thread");
  }
dengyihao's avatar
dengyihao 已提交
393 394 395 396 397
  if (!uv_is_closing((uv_handle_t*)req->data)) {
    uv_close((uv_handle_t*)req->data, uvFreeCb);
  } else {
    taosMemoryFree(req->data);
  }
wafwerar's avatar
wafwerar 已提交
398
  taosMemoryFree(req);
dengyihao's avatar
dengyihao 已提交
399
}
dengyihao's avatar
dengyihao 已提交
400

dengyihao's avatar
dengyihao 已提交
401
static int uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
dengyihao's avatar
dengyihao 已提交
402
  SSvrConn*  pConn = smsg->pConn;
dengyihao's avatar
formate  
dengyihao 已提交
403
  STransMsg* pMsg = &smsg->msg;
dengyihao's avatar
dengyihao 已提交
404 405 406 407 408
  if (pMsg->pCont == 0) {
    pMsg->pCont = (void*)rpcMallocCont(0);
    pMsg->contLen = 0;
  }
  STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
S
Shengliang Guan 已提交
409
  pHead->ahandle = (uint64_t)pMsg->info.ahandle;
dengyihao's avatar
dengyihao 已提交
410
  pHead->traceId = pMsg->info.traceId;
dengyihao's avatar
dengyihao 已提交
411
  pHead->hasEpSet = pMsg->info.hasEpSet;
dengyihao's avatar
dengyihao 已提交
412
  pHead->magicNum = htonl(TRANS_MAGIC_NUM);
dengyihao's avatar
dengyihao 已提交
413

dengyihao's avatar
dengyihao 已提交
414
  // handle invalid drop_task resp, TD-20098
dengyihao's avatar
dengyihao 已提交
415
  if (pConn->inType == TDMT_SCH_DROP_TASK && pMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID) {
dengyihao's avatar
dengyihao 已提交
416 417 418 419 420
    transQueuePop(&pConn->srvMsgs);
    destroySmsg(smsg);
    return -1;
  }

dengyihao's avatar
dengyihao 已提交
421
  if (pConn->status == ConnNormal) {
dengyihao's avatar
dengyihao 已提交
422
    pHead->msgType = (0 == pMsg->msgType ? pConn->inType + 1 : pMsg->msgType);
dengyihao's avatar
dengyihao 已提交
423
    if (smsg->type == Release) pHead->msgType = 0;
dengyihao's avatar
dengyihao 已提交
424
  } else {
dengyihao's avatar
dengyihao 已提交
425 426 427
    if (smsg->type == Release) {
      pHead->msgType = 0;
      pConn->status = ConnNormal;
dengyihao's avatar
dengyihao 已提交
428
      destroyConnRegArg(pConn);
dengyihao's avatar
dengyihao 已提交
429 430
      transUnrefSrvHandle(pConn);
    } else {
dengyihao's avatar
dengyihao 已提交
431
      // set up resp msg type
dengyihao's avatar
dengyihao 已提交
432
      pHead->msgType = (0 == pMsg->msgType ? pConn->inType + 1 : pMsg->msgType);
dengyihao's avatar
dengyihao 已提交
433
    }
dengyihao's avatar
dengyihao 已提交
434
  }
dengyihao's avatar
dengyihao 已提交
435

dengyihao's avatar
dengyihao 已提交
436
  pHead->release = smsg->type == Release ? 1 : 0;
dengyihao's avatar
dengyihao 已提交
437
  pHead->code = htonl(pMsg->code);
dengyihao's avatar
dengyihao 已提交
438
  pHead->msgLen = htonl(pMsg->contLen + sizeof(STransMsgHead));
dengyihao's avatar
dengyihao 已提交
439

dengyihao's avatar
dengyihao 已提交
440 441
  char*   msg = (char*)pHead;
  int32_t len = transMsgLenFromCont(pMsg->contLen);
dengyihao's avatar
dengyihao 已提交
442

dengyihao's avatar
dengyihao 已提交
443
  STrans* pTransInst = pConn->pTransInst;
dengyihao's avatar
dengyihao 已提交
444
  if (pTransInst->compressSize != -1 && pTransInst->compressSize < pMsg->contLen) {
dengyihao's avatar
dengyihao 已提交
445 446 447 448
    len = transCompressMsg(pMsg->pCont, pMsg->contLen) + sizeof(STransMsgHead);
    pHead->msgLen = (int32_t)htonl((uint32_t)len);
  }

dengyihao's avatar
dengyihao 已提交
449
  STraceId* trace = &pMsg->info.traceId;
dengyihao's avatar
dengyihao 已提交
450
  tGDebug("%s conn %p %s is sent to %s, local info:%s, len:%d", transLabel(pTransInst), pConn,
dengyihao's avatar
dengyihao 已提交
451
          TMSG_INFO(pHead->msgType), pConn->dst, pConn->src, len);
dengyihao's avatar
dengyihao 已提交
452

dengyihao's avatar
dengyihao 已提交
453
  wb->base = (char*)pHead;
dengyihao's avatar
dengyihao 已提交
454
  wb->len = len;
dengyihao's avatar
dengyihao 已提交
455
  return 0;
dengyihao's avatar
dengyihao 已提交
456
}
dengyihao's avatar
dengyihao 已提交
457

dengyihao's avatar
dengyihao 已提交
458
static FORCE_INLINE void uvStartSendRespImpl(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
459 460 461 462 463
  SSvrConn* pConn = smsg->pConn;
  if (pConn->broken) {
    return;
  }

dengyihao's avatar
dengyihao 已提交
464
  uv_buf_t wb;
dengyihao's avatar
dengyihao 已提交
465 466 467
  if (uvPrepareSendData(smsg, &wb) < 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
468

dengyihao's avatar
dengyihao 已提交
469
  transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
470
  uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
dengyihao's avatar
dengyihao 已提交
471
  uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb);
dengyihao's avatar
dengyihao 已提交
472
}
dengyihao's avatar
dengyihao 已提交
473
static void uvStartSendResp(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
474
  // impl
dengyihao's avatar
dengyihao 已提交
475
  SSvrConn* pConn = smsg->pConn;
dengyihao's avatar
dengyihao 已提交
476
  if (pConn->broken == true) {
dengyihao's avatar
dengyihao 已提交
477
    // persist by
dengyihao's avatar
dengyihao 已提交
478
    destroySmsg(smsg);
dengyihao's avatar
dengyihao 已提交
479 480 481
    transUnrefSrvHandle(pConn);
    return;
  }
dengyihao's avatar
dengyihao 已提交
482 483 484
  if (pConn->status == ConnNormal) {
    transUnrefSrvHandle(pConn);
  }
dengyihao's avatar
dengyihao 已提交
485

dengyihao's avatar
dengyihao 已提交
486
  if (!transQueuePush(&pConn->srvMsgs, smsg)) {
dengyihao's avatar
dengyihao 已提交
487 488
    return;
  }
dengyihao's avatar
dengyihao 已提交
489
  uvStartSendRespImpl(smsg);
dengyihao's avatar
dengyihao 已提交
490 491
  return;
}
dengyihao's avatar
dengyihao 已提交
492

dengyihao's avatar
dengyihao 已提交
493
static FORCE_INLINE void destroySmsg(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
494 495 496 497
  if (smsg == NULL) {
    return;
  }
  transFreeMsg(smsg->msg.pCont);
wafwerar's avatar
wafwerar 已提交
498
  taosMemoryFree(smsg);
dengyihao's avatar
dengyihao 已提交
499
}
dengyihao's avatar
dengyihao 已提交
500
static void destroyAllConn(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
501
  tTrace("thread %p destroy all conn ", pThrd);
dengyihao's avatar
fix bug  
dengyihao 已提交
502 503 504 505 506
  while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
    queue* h = QUEUE_HEAD(&pThrd->conn);
    QUEUE_REMOVE(h);
    QUEUE_INIT(h);

dengyihao's avatar
dengyihao 已提交
507
    SSvrConn* c = QUEUE_DATA(h, SSvrConn, queue);
dengyihao's avatar
dengyihao 已提交
508 509 510
    while (T_REF_VAL_GET(c) >= 2) {
      transUnrefSrvHandle(c);
    }
dengyihao's avatar
dengyihao 已提交
511
    transUnrefSrvHandle(c);
dengyihao's avatar
fix bug  
dengyihao 已提交
512 513
  }
}
dengyihao's avatar
dengyihao 已提交
514
void uvWorkerAsyncCb(uv_async_t* handle) {
dengyihao's avatar
dengyihao 已提交
515 516 517 518
  SAsyncItem* item = handle->data;
  SWorkThrd*  pThrd = item->pThrd;
  SSvrConn*   conn = NULL;
  queue       wq;
dengyihao's avatar
dengyihao 已提交
519

dengyihao's avatar
dengyihao 已提交
520
  // batch process to avoid to lock/unlock frequently
wafwerar's avatar
wafwerar 已提交
521
  taosThreadMutexLock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
522
  QUEUE_MOVE(&item->qmsg, &wq);
wafwerar's avatar
wafwerar 已提交
523
  taosThreadMutexUnlock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
524 525 526 527

  while (!QUEUE_IS_EMPTY(&wq)) {
    queue* head = QUEUE_HEAD(&wq);
    QUEUE_REMOVE(head);
dengyihao's avatar
dengyihao 已提交
528

dengyihao's avatar
dengyihao 已提交
529
    SSvrMsg* msg = QUEUE_DATA(head, SSvrMsg, q);
dengyihao's avatar
dengyihao 已提交
530
    if (msg == NULL) {
dengyihao's avatar
dengyihao 已提交
531
      tError("unexcept occurred, continue");
dengyihao's avatar
dengyihao 已提交
532
      continue;
dengyihao's avatar
dengyihao 已提交
533
    }
dengyihao's avatar
dengyihao 已提交
534

dengyihao's avatar
dengyihao 已提交
535
    // release handle to rpc init
dengyihao's avatar
dengyihao 已提交
536 537 538 539 540
    if (msg->type == Quit) {
      (*transAsyncHandle[msg->type])(msg, pThrd);
    } else {
      STransMsg transMsg = msg->msg;

S
Shengliang Guan 已提交
541 542
      SExHandle* exh1 = transMsg.info.handle;
      int64_t    refId = transMsg.info.refId;
dengyihao's avatar
dengyihao 已提交
543
      SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
544
      if (exh2 == NULL || exh1 != exh2) {
dengyihao's avatar
dengyihao 已提交
545
        tTrace("handle except msg %p, ignore it", exh1);
dengyihao's avatar
dengyihao 已提交
546
        transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
547 548 549 550
        destroySmsg(msg);
        continue;
      }
      msg->pConn = exh1->handle;
dengyihao's avatar
dengyihao 已提交
551
      transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
552
      (*transAsyncHandle[msg->type])(msg, pThrd);
dengyihao's avatar
dengyihao 已提交
553
    }
dengyihao's avatar
dengyihao 已提交
554 555
  }
}
dengyihao's avatar
dengyihao 已提交
556 557 558 559 560
static void uvWalkCb(uv_handle_t* handle, void* arg) {
  if (!uv_is_closing(handle)) {
    uv_close(handle, NULL);
  }
}
dengyihao's avatar
dengyihao 已提交
561 562 563 564
static void uvFreeCb(uv_handle_t* handle) {
  //
  taosMemoryFree(handle);
}
dengyihao's avatar
dengyihao 已提交
565

dengyihao's avatar
dengyihao 已提交
566 567
static void uvAcceptAsyncCb(uv_async_t* async) {
  SServerObj* srv = async->data;
dengyihao's avatar
dengyihao 已提交
568
  tDebug("close server port %d", srv->port);
dengyihao's avatar
dengyihao 已提交
569
  uv_walk(srv->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
570
}
dengyihao's avatar
dengyihao 已提交
571

dengyihao's avatar
dengyihao 已提交
572
static void uvShutDownCb(uv_shutdown_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
573
  if (status != 0) {
S
Shengliang Guan 已提交
574
    tDebug("conn failed to shut down:%s", uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
575
  }
dengyihao's avatar
dengyihao 已提交
576
  uv_close((uv_handle_t*)req->handle, uvDestroyConn);
wafwerar's avatar
wafwerar 已提交
577
  taosMemoryFree(req);
dengyihao's avatar
dengyihao 已提交
578
}
dengyihao's avatar
dengyihao 已提交
579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607
static bool uvRecvReleaseReq(SSvrConn* pConn, STransMsgHead* pHead) {
  if ((pHead)->release == 1 && (pHead->msgLen) == sizeof(*pHead)) {
    reallocConnRef(pConn);
    tTrace("conn %p received release request", pConn);

    STraceId traceId = pHead->traceId;
    pConn->status = ConnRelease;
    transClearBuffer(&pConn->readBuf);
    transFreeMsg(transContFromHead((char*)pHead));

    STransMsg tmsg = {.code = 0, .info.handle = (void*)pConn, .info.traceId = traceId, .info.ahandle = (void*)0x9527};
    SSvrMsg*  srvMsg = taosMemoryCalloc(1, sizeof(SSvrMsg));
    srvMsg->msg = tmsg;
    srvMsg->type = Release;
    srvMsg->pConn = pConn;
    if (!transQueuePush(&pConn->srvMsgs, srvMsg)) {
      return true;
    }
    if (pConn->regArg.init) {
      tTrace("conn %p release, notify server app", pConn);
      STrans* pTransInst = pConn->pTransInst;
      (*pTransInst->cfp)(pTransInst->parent, &(pConn->regArg.msg), NULL);
      memset(&pConn->regArg, 0, sizeof(pConn->regArg));
    }
    uvStartSendRespImpl(srvMsg);
    return true;
  }
  return false;
}
dengyihao's avatar
dengyihao 已提交
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653
static void uvPrepareCb(uv_prepare_t* handle) {
  // prepare callback
  SWorkThrd*  pThrd = handle->data;
  SAsyncPool* pool = pThrd->asyncPool;

  for (int i = 0; i < pool->nAsync; i++) {
    uv_async_t* async = &(pool->asyncs[i]);
    SAsyncItem* item = async->data;

    queue wq;
    taosThreadMutexLock(&item->mtx);
    QUEUE_MOVE(&item->qmsg, &wq);
    taosThreadMutexUnlock(&item->mtx);

    while (!QUEUE_IS_EMPTY(&wq)) {
      queue* head = QUEUE_HEAD(&wq);
      QUEUE_REMOVE(head);

      SSvrMsg* msg = QUEUE_DATA(head, SSvrMsg, q);
      if (msg == NULL) {
        tError("unexcept occurred, continue");
        continue;
      }
      // release handle to rpc init
      if (msg->type == Quit) {
        (*transAsyncHandle[msg->type])(msg, pThrd);
        continue;
      } else {
        STransMsg transMsg = msg->msg;

        SExHandle* exh1 = transMsg.info.handle;
        int64_t    refId = transMsg.info.refId;
        SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
        if (exh2 == NULL || exh1 != exh2) {
          tTrace("handle except msg %p, ignore it", exh1);
          transReleaseExHandle(transGetRefMgt(), refId);
          destroySmsg(msg);
          continue;
        }
        msg->pConn = exh1->handle;
        transReleaseExHandle(transGetRefMgt(), refId);
        (*transAsyncHandle[msg->type])(msg, pThrd);
      }
    }
  }
}
dengyihao's avatar
dengyihao 已提交
654

dengyihao's avatar
dengyihao 已提交
655 656 657
static void uvWorkDoTask(uv_work_t* req) {
  // doing time-consumeing task
  // only auth conn currently, add more func later
dengyihao's avatar
dengyihao 已提交
658
  tTrace("conn %p start to be processed in BG Thread", req->data);
dengyihao's avatar
dengyihao 已提交
659 660 661 662 663
  return;
}

static void uvWorkAfterTask(uv_work_t* req, int status) {
  if (status != 0) {
dengyihao's avatar
dengyihao 已提交
664
    tTrace("conn %p failed to processed ", req->data);
dengyihao's avatar
dengyihao 已提交
665 666 667 668
  }
  // Done time-consumeing task
  // add more func later
  // this func called in main loop
dengyihao's avatar
dengyihao 已提交
669
  tTrace("conn %p already processed ", req->data);
dengyihao's avatar
dengyihao 已提交
670 671 672
  taosMemoryFree(req);
}

dengyihao's avatar
dengyihao 已提交
673 674 675 676 677 678
void uvOnAcceptCb(uv_stream_t* stream, int status) {
  if (status == -1) {
    return;
  }
  SServerObj* pObj = container_of(stream, SServerObj, server);

wafwerar's avatar
wafwerar 已提交
679
  uv_tcp_t* cli = (uv_tcp_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
680 681 682
  uv_tcp_init(pObj->loop, cli);

  if (uv_accept(stream, (uv_stream_t*)cli) == 0) {
dengyihao's avatar
dengyihao 已提交
683
#if defined(WINDOWS) || defined(DARWIN)
wafwerar's avatar
wafwerar 已提交
684
    if (pObj->numOfWorkerReady < pObj->numOfThreads) {
dengyihao's avatar
dengyihao 已提交
685 686
      tError("worker-threads are not ready for all, need %d instead of %d.", pObj->numOfThreads,
             pObj->numOfWorkerReady);
wafwerar's avatar
wafwerar 已提交
687 688 689
      uv_close((uv_handle_t*)cli, NULL);
      return;
    }
dengyihao's avatar
dengyihao 已提交
690
#endif
dengyihao's avatar
dengyihao 已提交
691

wafwerar's avatar
wafwerar 已提交
692
    uv_write_t* wr = (uv_write_t*)taosMemoryMalloc(sizeof(uv_write_t));
dengyihao's avatar
dengyihao 已提交
693
    wr->data = cli;
dengyihao's avatar
dengyihao 已提交
694 695 696
    uv_buf_t buf = uv_buf_init((char*)notify, strlen(notify));

    pObj->workerIdx = (pObj->workerIdx + 1) % pObj->numOfThreads;
dengyihao's avatar
dengyihao 已提交
697

dengyihao's avatar
dengyihao 已提交
698
    tTrace("new connection accepted by main server, dispatch to %dth worker-thread", pObj->workerIdx);
dengyihao's avatar
dengyihao 已提交
699

dengyihao's avatar
dengyihao 已提交
700
    uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb);
dengyihao's avatar
dengyihao 已提交
701
  } else {
dengyihao's avatar
dengyihao 已提交
702 703 704 705 706
    if (!uv_is_closing((uv_handle_t*)cli)) {
      uv_close((uv_handle_t*)cli, NULL);
    } else {
      taosMemoryFree(cli);
    }
dengyihao's avatar
dengyihao 已提交
707 708 709
  }
}
void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
710
  tTrace("connection coming");
dengyihao's avatar
dengyihao 已提交
711 712 713 714 715
  if (nread < 0) {
    if (nread != UV_EOF) {
      tError("read error %s", uv_err_name(nread));
    }
    // TODO(log other failure reason)
S
Shengliang Guan 已提交
716
    tWarn("failed to create connect:%p", q);
dengyihao's avatar
dengyihao 已提交
717 718
    taosMemoryFree(buf->base);
    uv_close((uv_handle_t*)q, NULL);
dengyihao's avatar
dengyihao 已提交
719 720 721
    return;
  }
  // free memory allocated by
dengyihao's avatar
dengyihao 已提交
722 723
  ASSERTS(nread == strlen(notify), "trans-svr mem corrupted");
  ASSERTS(buf->base[0] == notify[0], "trans-svr mem corrupted");
wafwerar's avatar
wafwerar 已提交
724
  taosMemoryFree(buf->base);
dengyihao's avatar
dengyihao 已提交
725

dengyihao's avatar
dengyihao 已提交
726
  SWorkThrd* pThrd = q->data;
dengyihao's avatar
dengyihao 已提交
727 728 729 730 731 732 733 734 735

  uv_pipe_t* pipe = (uv_pipe_t*)q;
  if (!uv_pipe_pending_count(pipe)) {
    tError("No pending count");
    return;
  }

  uv_handle_type pending = uv_pipe_pending_type(pipe);

dengyihao's avatar
dengyihao 已提交
736
  SSvrConn* pConn = createConn(pThrd);
737

dengyihao's avatar
dengyihao 已提交
738
  pConn->pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
739
  /* init conn timer*/
dengyihao's avatar
dengyihao 已提交
740 741
  // uv_timer_init(pThrd->loop, &pConn->pTimer);
  // pConn->pTimer.data = pConn;
dengyihao's avatar
dengyihao 已提交
742 743 744 745

  pConn->hostThrd = pThrd;

  // init client handle
wafwerar's avatar
wafwerar 已提交
746
  pConn->pTcp = (uv_tcp_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
747 748 749
  uv_tcp_init(pThrd->loop, pConn->pTcp);
  pConn->pTcp->data = pConn;

dengyihao's avatar
dengyihao 已提交
750 751
  transSetConnOption((uv_tcp_t*)pConn->pTcp);

dengyihao's avatar
dengyihao 已提交
752 753 754
  if (uv_accept(q, (uv_stream_t*)(pConn->pTcp)) == 0) {
    uv_os_fd_t fd;
    uv_fileno((const uv_handle_t*)pConn->pTcp, &fd);
S
Shengliang Guan 已提交
755
    tTrace("conn %p created, fd:%d", pConn, fd);
dengyihao's avatar
dengyihao 已提交
756

dengyihao's avatar
dengyihao 已提交
757 758 759
    struct sockaddr peername, sockname;
    int             addrlen = sizeof(peername);
    if (0 != uv_tcp_getpeername(pConn->pTcp, (struct sockaddr*)&peername, &addrlen)) {
dengyihao's avatar
dengyihao 已提交
760
      tError("conn %p failed to get peer info", pConn);
dengyihao's avatar
dengyihao 已提交
761
      transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
762
      return;
dengyihao's avatar
dengyihao 已提交
763
    }
dengyihao's avatar
dengyihao 已提交
764
    transSockInfo2Str(&peername, pConn->dst);
dengyihao's avatar
dengyihao 已提交
765

dengyihao's avatar
dengyihao 已提交
766 767
    addrlen = sizeof(sockname);
    if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&sockname, &addrlen)) {
dengyihao's avatar
dengyihao 已提交
768
      tError("conn %p failed to get local info", pConn);
dengyihao's avatar
dengyihao 已提交
769
      transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
770 771
      return;
    }
dengyihao's avatar
dengyihao 已提交
772
    transSockInfo2Str(&sockname, pConn->src);
dengyihao's avatar
dengyihao 已提交
773

dengyihao's avatar
dengyihao 已提交
774
    struct sockaddr_in addr = *(struct sockaddr_in*)&peername;
dengyihao's avatar
dengyihao 已提交
775 776
    pConn->clientIp = addr.sin_addr.s_addr;
    pConn->port = ntohs(addr.sin_port);
dengyihao's avatar
dengyihao 已提交
777

dengyihao's avatar
dengyihao 已提交
778
    uv_read_start((uv_stream_t*)(pConn->pTcp), uvAllocRecvBufferCb, uvOnRecvCb);
dengyihao's avatar
dengyihao 已提交
779

dengyihao's avatar
dengyihao 已提交
780
  } else {
dengyihao's avatar
dengyihao 已提交
781
    tDebug("failed to create new connection");
dengyihao's avatar
dengyihao 已提交
782
    transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
783 784 785
  }
}

786
void* transAcceptThread(void* arg) {
dengyihao's avatar
dengyihao 已提交
787
  // opt
dengyihao's avatar
dengyihao 已提交
788
  setThreadName("trans-accept");
dengyihao's avatar
dengyihao 已提交
789 790
  SServerObj* srv = (SServerObj*)arg;
  uv_run(srv->loop, UV_RUN_DEFAULT);
791 792

  return NULL;
dengyihao's avatar
dengyihao 已提交
793
}
dengyihao's avatar
dengyihao 已提交
794
void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
wafwerar's avatar
wafwerar 已提交
795 796 797
  if (status != 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
798
  SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
wafwerar's avatar
wafwerar 已提交
799 800
  uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
}
dengyihao's avatar
dengyihao 已提交
801
static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
wafwerar's avatar
wafwerar 已提交
802
  pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
803 804 805
  if (0 != uv_loop_init(pThrd->loop)) {
    return false;
  }
dengyihao's avatar
dengyihao 已提交
806

dengyihao's avatar
dengyihao 已提交
807
#if defined(WINDOWS) || defined(DARWIN)
dengyihao's avatar
dengyihao 已提交
808
  uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
dengyihao's avatar
dengyihao 已提交
809 810 811 812
#else
  uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
  uv_pipe_open(pThrd->pipe, pThrd->fd);
#endif
dengyihao's avatar
dengyihao 已提交
813 814 815

  pThrd->pipe->data = pThrd;

dengyihao's avatar
dengyihao 已提交
816
  QUEUE_INIT(&pThrd->msg);
dengyihao's avatar
dengyihao 已提交
817

dengyihao's avatar
dengyihao 已提交
818 819 820 821 822
  pThrd->prepare = taosMemoryCalloc(1, sizeof(uv_prepare_t));
  uv_prepare_init(pThrd->loop, pThrd->prepare);
  uv_prepare_start(pThrd->prepare, uvPrepareCb);
  pThrd->prepare->data = pThrd;

dengyihao's avatar
fix bug  
dengyihao 已提交
823 824 825
  // conn set
  QUEUE_INIT(&pThrd->conn);

dengyihao's avatar
dengyihao 已提交
826
  pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 8, pThrd, uvWorkerAsyncCb);
dengyihao's avatar
dengyihao 已提交
827
#if defined(WINDOWS) || defined(DARWIN)
wafwerar's avatar
wafwerar 已提交
828
  uv_pipe_connect(&pThrd->connect_req, pThrd->pipe, pipeName, uvOnPipeConnectionCb);
dengyihao's avatar
dengyihao 已提交
829 830 831
#else
  uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
#endif
dengyihao's avatar
dengyihao 已提交
832 833 834 835 836 837 838 839 840
  return true;
}

static bool addHandleToAcceptloop(void* arg) {
  // impl later
  SServerObj* srv = arg;

  int err = 0;
  if ((err = uv_tcp_init(srv->loop, &srv->server)) != 0) {
S
Shengliang Guan 已提交
841
    tError("failed to init accept server:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
842 843 844
    return false;
  }

dengyihao's avatar
dengyihao 已提交
845
  // register an async here to quit server gracefully
wafwerar's avatar
wafwerar 已提交
846
  srv->pAcceptAsync = taosMemoryCalloc(1, sizeof(uv_async_t));
dengyihao's avatar
dengyihao 已提交
847 848
  uv_async_init(srv->loop, srv->pAcceptAsync, uvAcceptAsyncCb);
  srv->pAcceptAsync->data = srv;
dengyihao's avatar
dengyihao 已提交
849

dengyihao's avatar
dengyihao 已提交
850
  struct sockaddr_in bind_addr;
dengyihao's avatar
dengyihao 已提交
851 852
  uv_ip4_addr("0.0.0.0", srv->port, &bind_addr);
  if ((err = uv_tcp_bind(&srv->server, (const struct sockaddr*)&bind_addr, 0)) != 0) {
S
Shengliang Guan 已提交
853
    tError("failed to bind:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
854 855
    return false;
  }
dengyihao's avatar
dengyihao 已提交
856
  if ((err = uv_listen((uv_stream_t*)&srv->server, 4096 * 2, uvOnAcceptCb)) != 0) {
S
Shengliang Guan 已提交
857
    tError("failed to listen:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
858
    terrno = TSDB_CODE_RPC_PORT_EADDRINUSE;
dengyihao's avatar
dengyihao 已提交
859 860 861
    return false;
  }
  return true;
dengyihao's avatar
dengyihao 已提交
862
}
863
void* transWorkerThread(void* arg) {
G
gccgdb1234 已提交
864
  setThreadName("trans-svr-work");
dengyihao's avatar
dengyihao 已提交
865
  SWorkThrd* pThrd = (SWorkThrd*)arg;
dengyihao's avatar
dengyihao 已提交
866
  uv_run(pThrd->loop, UV_RUN_DEFAULT);
867 868

  return NULL;
dengyihao's avatar
dengyihao 已提交
869 870
}

dengyihao's avatar
dengyihao 已提交
871
static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
dengyihao's avatar
dengyihao 已提交
872
  SWorkThrd* pThrd = hThrd;
dengyihao's avatar
fix bug  
dengyihao 已提交
873

dengyihao's avatar
dengyihao 已提交
874
  SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn));
dengyihao's avatar
dengyihao 已提交
875

dengyihao's avatar
dengyihao 已提交
876
  transReqQueueInit(&pConn->wreqQueue);
dengyihao's avatar
fix bug  
dengyihao 已提交
877 878 879
  QUEUE_INIT(&pConn->queue);

  QUEUE_PUSH(&pThrd->conn, &pConn->queue);
dengyihao's avatar
dengyihao 已提交
880 881

  transQueueInit(&pConn->srvMsgs, NULL);
dengyihao's avatar
dengyihao 已提交
882

dengyihao's avatar
dengyihao 已提交
883
  memset(&pConn->regArg, 0, sizeof(pConn->regArg));
dengyihao's avatar
dengyihao 已提交
884
  pConn->broken = false;
dengyihao's avatar
dengyihao 已提交
885
  pConn->status = ConnNormal;
dengyihao's avatar
opt rpc  
dengyihao 已提交
886
  transInitBuffer(&pConn->readBuf);
dengyihao's avatar
dengyihao 已提交
887

dengyihao's avatar
dengyihao 已提交
888 889 890
  SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
  exh->handle = pConn;
  exh->pThrd = pThrd;
dengyihao's avatar
dengyihao 已提交
891 892
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
893

dengyihao's avatar
dengyihao 已提交
894
  STrans* pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
895
  pConn->refId = exh->refId;
dengyihao's avatar
dengyihao 已提交
896
  transRefSrvHandle(pConn);
S
Shengliang Guan 已提交
897
  tTrace("%s handle %p, conn %p created, refId:%" PRId64, transLabel(pTransInst), exh, pConn, pConn->refId);
dengyihao's avatar
dengyihao 已提交
898 899
  return pConn;
}
dengyihao's avatar
dengyihao 已提交
900

dengyihao's avatar
dengyihao 已提交
901
static FORCE_INLINE void destroyConn(SSvrConn* conn, bool clear) {
dengyihao's avatar
dengyihao 已提交
902 903 904
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
905

906
  if (clear) {
dengyihao's avatar
dengyihao 已提交
907
    if (!uv_is_closing((uv_handle_t*)conn->pTcp)) {
dengyihao's avatar
dengyihao 已提交
908 909 910
      tTrace("conn %p to be destroyed", conn);
      uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn);
    }
911
  }
dengyihao's avatar
dengyihao 已提交
912
}
dengyihao's avatar
dengyihao 已提交
913
static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
914 915 916 917 918
  if (conn->regArg.init == 1) {
    transFreeMsg(conn->regArg.msg.pCont);
    conn->regArg.init = 0;
  }
}
dengyihao's avatar
dengyihao 已提交
919
static int reallocConnRef(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
920 921
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
922 923 924 925
  // avoid app continue to send msg on invalid handle
  SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
  exh->handle = conn;
  exh->pThrd = conn->hostThrd;
dengyihao's avatar
dengyihao 已提交
926 927
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
928 929 930 931
  conn->refId = exh->refId;

  return 0;
}
dengyihao's avatar
dengyihao 已提交
932
static void uvDestroyConn(uv_handle_t* handle) {
dengyihao's avatar
dengyihao 已提交
933
  SSvrConn* conn = handle->data;
dengyihao's avatar
dengyihao 已提交
934

dengyihao's avatar
dengyihao 已提交
935 936 937
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
938
  SWorkThrd* thrd = conn->hostThrd;
dengyihao's avatar
dengyihao 已提交
939

dengyihao's avatar
dengyihao 已提交
940 941
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
942

dengyihao's avatar
dengyihao 已提交
943 944
  STrans* pTransInst = thrd->pTransInst;
  tDebug("%s conn %p destroy", transLabel(pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
945

dengyihao's avatar
dengyihao 已提交
946 947 948 949
  for (int i = 0; i < transQueueSize(&conn->srvMsgs); i++) {
    SSvrMsg* msg = transQueueGet(&conn->srvMsgs, i);
    destroySmsg(msg);
  }
dengyihao's avatar
dengyihao 已提交
950
  transQueueDestroy(&conn->srvMsgs);
dengyihao's avatar
dengyihao 已提交
951
  transReqQueueClear(&conn->wreqQueue);
dengyihao's avatar
dengyihao 已提交
952

dengyihao's avatar
dengyihao 已提交
953
  QUEUE_REMOVE(&conn->queue);
wafwerar's avatar
wafwerar 已提交
954
  taosMemoryFree(conn->pTcp);
dengyihao's avatar
dengyihao 已提交
955
  destroyConnRegArg(conn);
dengyihao's avatar
dengyihao 已提交
956
  transDestroyBuffer(&conn->readBuf);
dengyihao's avatar
dengyihao 已提交
957
  taosMemoryFree(conn);
dengyihao's avatar
dengyihao 已提交
958

dengyihao's avatar
dengyihao 已提交
959
  if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
dengyihao's avatar
dengyihao 已提交
960
    tTrace("work thread quit");
dengyihao's avatar
dengyihao 已提交
961
    uv_walk(thrd->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
962
  }
dengyihao's avatar
dengyihao 已提交
963
}
wafwerar's avatar
wafwerar 已提交
964
static void uvPipeListenCb(uv_stream_t* handle, int status) {
dengyihao's avatar
dengyihao 已提交
965
  if (status != 0) {
dengyihao's avatar
dengyihao 已提交
966
    tError("server failed to init pipe, errmsg: %s", uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
967 968
    return;
  }
wafwerar's avatar
wafwerar 已提交
969 970

  SServerObj* srv = container_of(handle, SServerObj, pipeListen);
dengyihao's avatar
dengyihao 已提交
971
  uv_pipe_t*  pipe = &(srv->pipe[srv->numOfWorkerReady][0]);
wafwerar's avatar
wafwerar 已提交
972

dengyihao's avatar
dengyihao 已提交
973
  int ret = uv_pipe_init(srv->loop, pipe, 1);
dengyihao's avatar
dengyihao 已提交
974 975
  ASSERTS(ret == 0, "trans-svr failed to init pipe");
  if (ret != 0) return;
wafwerar's avatar
wafwerar 已提交
976

dengyihao's avatar
dengyihao 已提交
977
  ret = uv_accept((uv_stream_t*)&srv->pipeListen, (uv_stream_t*)pipe);
dengyihao's avatar
dengyihao 已提交
978 979
  ASSERTS(ret == 0, "trans-svr failed to accept pipe msg");
  if (ret != 0) return;
dengyihao's avatar
dengyihao 已提交
980 981

  ret = uv_is_readable((uv_stream_t*)pipe);
dengyihao's avatar
dengyihao 已提交
982 983
  ASSERTS(ret == 1, "trans-svr pipe status corrupted");
  if (ret != 1) return;
wafwerar's avatar
wafwerar 已提交
984

dengyihao's avatar
dengyihao 已提交
985
  ret = uv_is_writable((uv_stream_t*)pipe);
dengyihao's avatar
dengyihao 已提交
986
  ASSERTS(ret == 1, "trans-svr pipe status corrupted");
dengyihao's avatar
dengyihao 已提交
987
  if (ret != 1) return;
wafwerar's avatar
wafwerar 已提交
988

dengyihao's avatar
dengyihao 已提交
989
  ret = uv_is_closing((uv_handle_t*)pipe);
dengyihao's avatar
dengyihao 已提交
990 991
  ASSERTS(ret == 0, "trans-svr pipe status corrupted");
  if (ret != 0) return;
dengyihao's avatar
dengyihao 已提交
992 993

  srv->numOfWorkerReady++;
wafwerar's avatar
wafwerar 已提交
994
}
dengyihao's avatar
dengyihao 已提交
995

U
ubuntu 已提交
996
void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
wafwerar's avatar
wafwerar 已提交
997 998
  SServerObj* srv = taosMemoryCalloc(1, sizeof(SServerObj));
  srv->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
999 1000
  srv->numOfThreads = numOfThreads;
  srv->workerIdx = 0;
wafwerar's avatar
wafwerar 已提交
1001
  srv->numOfWorkerReady = 0;
dengyihao's avatar
dengyihao 已提交
1002
  srv->pThreadObj = (SWorkThrd**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrd*));
wafwerar's avatar
wafwerar 已提交
1003
  srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*));
dengyihao's avatar
dengyihao 已提交
1004 1005 1006 1007
  srv->ip = ip;
  srv->port = port;
  uv_loop_init(srv->loop);

dengyihao's avatar
dengyihao 已提交
1008
  char pipeName[PATH_MAX];
1009 1010 1011 1012 1013 1014 1015

  if (false == taosValidIpAndPort(srv->ip, srv->port)) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    tError("invalid ip/port, %d:%d, reason:%s", srv->ip, srv->port, terrstr());
    goto End;
  }

dengyihao's avatar
dengyihao 已提交
1016
#if defined(WINDOWS) || defined(DARWIN)
dengyihao's avatar
dengyihao 已提交
1017
  int ret = uv_pipe_init(srv->loop, &srv->pipeListen, 0);
dengyihao's avatar
dengyihao 已提交
1018 1019 1020 1021
  if (ret != 0) {
    tError("failed to init pipe, errmsg: %s", uv_err_name(ret));
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1022
#if defined(WINDOWS)
dengyihao's avatar
dengyihao 已提交
1023
  snprintf(pipeName, sizeof(pipeName), "\\\\?\\pipe\\trans.rpc.%d-%" PRIu64, taosSafeRand(), GetCurrentProcessId());
dengyihao's avatar
dengyihao 已提交
1024 1025 1026 1027
#elif defined(DARWIN)
  snprintf(pipeName, sizeof(pipeName), "%s%spipe.trans.rpc.%08d-%" PRIu64, tsTempDir, TD_DIRSEP, taosSafeRand(),
           taosGetSelfPthreadId());
#endif
dengyihao's avatar
dengyihao 已提交
1028

dengyihao's avatar
dengyihao 已提交
1029
  ret = uv_pipe_bind(&srv->pipeListen, pipeName);
dengyihao's avatar
dengyihao 已提交
1030 1031 1032 1033
  if (ret != 0) {
    tError("failed to bind pipe, errmsg: %s", uv_err_name(ret));
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1034 1035

  ret = uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb);
dengyihao's avatar
dengyihao 已提交
1036 1037 1038 1039
  if (ret != 0) {
    tError("failed to listen pipe, errmsg: %s", uv_err_name(ret));
    goto End;
  }
wafwerar's avatar
wafwerar 已提交
1040

dengyihao's avatar
dengyihao 已提交
1041
  for (int i = 0; i < srv->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
1042
    SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));
dengyihao's avatar
dengyihao 已提交
1043
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
1044
    thrd->quit = false;
dengyihao's avatar
dengyihao 已提交
1045
    srv->pThreadObj[i] = thrd;
dengyihao's avatar
dengyihao 已提交
1046
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
1047

wafwerar's avatar
wafwerar 已提交
1048
    srv->pipe[i] = (uv_pipe_t*)taosMemoryCalloc(2, sizeof(uv_pipe_t));
dengyihao's avatar
dengyihao 已提交
1049
    thrd->pipe = &(srv->pipe[i][1]);  // init read
dengyihao's avatar
dengyihao 已提交
1050

dengyihao's avatar
dengyihao 已提交
1051
    if (false == addHandleToWorkloop(thrd, pipeName)) {
dengyihao's avatar
dengyihao 已提交
1052 1053
      goto End;
    }
dengyihao's avatar
dengyihao 已提交
1054

dengyihao's avatar
dengyihao 已提交
1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076
    int err = taosThreadCreate(&(thrd->thread), NULL, transWorkerThread, (void*)(thrd));
    if (err == 0) {
      tDebug("success to create worker-thread:%d", i);
    } else {
      // TODO: clear all other resource later
      tError("failed to create worker-thread:%d", i);
      goto End;
    }
  }
#else

  for (int i = 0; i < srv->numOfThreads; i++) {
    SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));

    thrd->pTransInst = shandle;
    thrd->quit = false;
    thrd->pTransInst = shandle;

    srv->pipe[i] = (uv_pipe_t*)taosMemoryCalloc(2, sizeof(uv_pipe_t));
    srv->pThreadObj[i] = thrd;

    uv_os_sock_t fds[2];
dengyihao's avatar
dengyihao 已提交
1077
    if (uv_socketpair(SOCK_STREAM, 0, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) {
dengyihao's avatar
dengyihao 已提交
1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090
      goto End;
    }

    uv_pipe_init(srv->loop, &(srv->pipe[i][0]), 1);
    uv_pipe_open(&(srv->pipe[i][0]), fds[1]);

    thrd->pipe = &(srv->pipe[i][1]);  // init read
    thrd->fd = fds[0];

    if (false == addHandleToWorkloop(thrd, pipeName)) {
      goto End;
    }

1091
    int err = taosThreadCreate(&(thrd->thread), NULL, transWorkerThread, (void*)(thrd));
dengyihao's avatar
dengyihao 已提交
1092
    if (err == 0) {
S
Shengliang Guan 已提交
1093
      tDebug("success to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
1094 1095
    } else {
      // TODO: clear all other resource later
S
Shengliang Guan 已提交
1096
      tError("failed to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
1097
      goto End;
dengyihao's avatar
dengyihao 已提交
1098 1099
    }
  }
dengyihao's avatar
dengyihao 已提交
1100
#endif
dengyihao's avatar
dengyihao 已提交
1101

dengyihao's avatar
dengyihao 已提交
1102 1103 1104
  if (false == addHandleToAcceptloop(srv)) {
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1105

1106
  int err = taosThreadCreate(&srv->thread, NULL, transAcceptThread, (void*)srv);
dengyihao's avatar
dengyihao 已提交
1107 1108 1109
  if (err == 0) {
    tDebug("success to create accept-thread");
  } else {
dengyihao's avatar
dengyihao 已提交
1110 1111
    tError("failed  to create accept-thread");
    goto End;
dengyihao's avatar
dengyihao 已提交
1112 1113
    // clear all resource later
  }
dengyihao's avatar
dengyihao 已提交
1114

dengyihao's avatar
dengyihao 已提交
1115
  srv->inited = true;
dengyihao's avatar
dengyihao 已提交
1116
  return srv;
dengyihao's avatar
dengyihao 已提交
1117
End:
U
ubuntu 已提交
1118
  transCloseServer(srv);
dengyihao's avatar
dengyihao 已提交
1119 1120
  return NULL;
}
dengyihao's avatar
dengyihao 已提交
1121

dengyihao's avatar
dengyihao 已提交
1122
void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
1123
  thrd->quit = true;
dengyihao's avatar
dengyihao 已提交
1124
  if (QUEUE_IS_EMPTY(&thrd->conn)) {
dengyihao's avatar
dengyihao 已提交
1125
    uv_walk(thrd->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
1126 1127 1128
  } else {
    destroyAllConn(thrd);
  }
wafwerar's avatar
wafwerar 已提交
1129
  taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
1130
}
dengyihao's avatar
dengyihao 已提交
1131
void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
1132
  SSvrConn* conn = msg->pConn;
dengyihao's avatar
dengyihao 已提交
1133
  if (conn->status == ConnAcquire) {
dengyihao's avatar
dengyihao 已提交
1134
    reallocConnRef(conn);
dengyihao's avatar
dengyihao 已提交
1135
    if (!transQueuePush(&conn->srvMsgs, msg)) {
dengyihao's avatar
dengyihao 已提交
1136
      return;
dengyihao's avatar
dengyihao 已提交
1137
    }
dengyihao's avatar
dengyihao 已提交
1138
    uvStartSendRespImpl(msg);
dengyihao's avatar
dengyihao 已提交
1139
    return;
dengyihao's avatar
dengyihao 已提交
1140
  } else if (conn->status == ConnRelease || conn->status == ConnNormal) {
dengyihao's avatar
dengyihao 已提交
1141
    tDebug("%s conn %p already released, ignore release-msg", transLabel(thrd->pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
1142
  }
dengyihao's avatar
dengyihao 已提交
1143
  destroySmsg(msg);
dengyihao's avatar
dengyihao 已提交
1144
}
dengyihao's avatar
dengyihao 已提交
1145
void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
1146
  // send msg to client
dengyihao's avatar
dengyihao 已提交
1147
  tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn);
dengyihao's avatar
dengyihao 已提交
1148 1149
  uvStartSendResp(msg);
}
dengyihao's avatar
dengyihao 已提交
1150
void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
1151
  SSvrConn* conn = msg->pConn;
dengyihao's avatar
dengyihao 已提交
1152
  tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
1153
  if (conn->status == ConnAcquire) {
dengyihao's avatar
dengyihao 已提交
1154
    if (!transQueuePush(&conn->srvMsgs, msg)) {
dengyihao's avatar
dengyihao 已提交
1155 1156
      return;
    }
dengyihao's avatar
dengyihao 已提交
1157
    transQueuePop(&conn->srvMsgs);
dengyihao's avatar
dengyihao 已提交
1158 1159 1160
    conn->regArg.notifyCount = 0;
    conn->regArg.init = 1;
    conn->regArg.msg = msg->msg;
dengyihao's avatar
dengyihao 已提交
1161
    tDebug("conn %p register brokenlink callback succ", conn);
dengyihao's avatar
dengyihao 已提交
1162 1163 1164 1165 1166 1167

    if (conn->broken) {
      STrans* pTransInst = conn->pTransInst;
      (*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
      memset(&conn->regArg, 0, sizeof(conn->regArg));
    }
wafwerar's avatar
wafwerar 已提交
1168
    taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
1169 1170
  }
}
dengyihao's avatar
dengyihao 已提交
1171
void destroyWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
1172 1173 1174
  if (pThrd == NULL) {
    return;
  }
wafwerar's avatar
wafwerar 已提交
1175
  taosThreadJoin(pThrd->thread, NULL);
dengyihao's avatar
dengyihao 已提交
1176
  SRV_RELEASE_UV(pThrd->loop);
dengyihao's avatar
dengyihao 已提交
1177
  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg);
dengyihao's avatar
dengyihao 已提交
1178
  transAsyncPoolDestroy(pThrd->asyncPool);
dengyihao's avatar
dengyihao 已提交
1179
  taosMemoryFree(pThrd->prepare);
dengyihao's avatar
dengyihao 已提交
1180
  taosMemoryFree(pThrd->loop);
wafwerar's avatar
wafwerar 已提交
1181
  taosMemoryFree(pThrd);
dengyihao's avatar
dengyihao 已提交
1182
}
dengyihao's avatar
dengyihao 已提交
1183
void sendQuitToWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
1184
  SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
dengyihao's avatar
dengyihao 已提交
1185
  msg->type = Quit;
dengyihao's avatar
dengyihao 已提交
1186
  tDebug("server send quit msg to work thread");
dengyihao's avatar
dengyihao 已提交
1187
  transAsyncSend(pThrd->asyncPool, &msg->q);
dengyihao's avatar
dengyihao 已提交
1188 1189
}

U
ubuntu 已提交
1190
void transCloseServer(void* arg) {
dengyihao's avatar
dengyihao 已提交
1191 1192
  // impl later
  SServerObj* srv = arg;
dengyihao's avatar
dengyihao 已提交
1193

dengyihao's avatar
dengyihao 已提交
1194
  if (srv->inited) {
1195
    tDebug("send quit msg to accept thread");
dengyihao's avatar
dengyihao 已提交
1196 1197
    uv_async_send(srv->pAcceptAsync);
    taosThreadJoin(srv->thread, NULL);
dengyihao's avatar
dengyihao 已提交
1198
    SRV_RELEASE_UV(srv->loop);
dengyihao's avatar
dengyihao 已提交
1199

dengyihao's avatar
dengyihao 已提交
1200 1201 1202 1203
    for (int i = 0; i < srv->numOfThreads; i++) {
      sendQuitToWorkThrd(srv->pThreadObj[i]);
      destroyWorkThrd(srv->pThreadObj[i]);
    }
dengyihao's avatar
dengyihao 已提交
1204 1205
  } else {
    uv_loop_close(srv->loop);
dengyihao's avatar
dengyihao 已提交
1206 1207
  }

wafwerar's avatar
wafwerar 已提交
1208 1209 1210
  taosMemoryFree(srv->pThreadObj);
  taosMemoryFree(srv->pAcceptAsync);
  taosMemoryFree(srv->loop);
dengyihao's avatar
dengyihao 已提交
1211 1212

  for (int i = 0; i < srv->numOfThreads; i++) {
wafwerar's avatar
wafwerar 已提交
1213
    taosMemoryFree(srv->pipe[i]);
dengyihao's avatar
dengyihao 已提交
1214
  }
wafwerar's avatar
wafwerar 已提交
1215
  taosMemoryFree(srv->pipe);
dengyihao's avatar
dengyihao 已提交
1216

wafwerar's avatar
wafwerar 已提交
1217
  taosMemoryFree(srv);
dengyihao's avatar
dengyihao 已提交
1218
}
dengyihao's avatar
dengyihao 已提交
1219

dengyihao's avatar
dengyihao 已提交
1220 1221 1222 1223
void transRefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1224
  int ref = T_REF_INC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1225
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1226 1227 1228 1229 1230 1231
}

void transUnrefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1232
  int ref = T_REF_DEC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1233
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1234
  if (ref == 0) {
dengyihao's avatar
dengyihao 已提交
1235
    destroyConn((SSvrConn*)handle, true);
dengyihao's avatar
dengyihao 已提交
1236 1237
  }
}
dengyihao's avatar
dengyihao 已提交
1238

dengyihao's avatar
dengyihao 已提交
1239
int transReleaseSrvHandle(void* handle) {
dengyihao's avatar
dengyihao 已提交
1240 1241 1242
  SRpcHandleInfo* info = handle;
  SExHandle*      exh = info->handle;
  int64_t         refId = info->refId;
dengyihao's avatar
dengyihao 已提交
1243

dengyihao's avatar
dengyihao 已提交
1244
  ASYNC_CHECK_HANDLE(exh, refId);
dengyihao's avatar
dengyihao 已提交
1245

dengyihao's avatar
dengyihao 已提交
1246
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1247 1248
  ASYNC_ERR_JRET(pThrd);

S
Shengliang Guan 已提交
1249
  STransMsg tmsg = {.code = 0, .info.handle = exh, .info.ahandle = NULL, .info.refId = refId};
dengyihao's avatar
dengyihao 已提交
1250

dengyihao's avatar
dengyihao 已提交
1251 1252 1253
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Release;
dengyihao's avatar
dengyihao 已提交
1254

dengyihao's avatar
dengyihao 已提交
1255
  tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
1256 1257 1258
  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
    destroySmsg(m);
  }
H
Haojun Liao 已提交
1259

dengyihao's avatar
dengyihao 已提交
1260
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1261
  return 0;
dengyihao's avatar
dengyihao 已提交
1262
_return1:
dengyihao's avatar
dengyihao 已提交
1263
  tDebug("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1264
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1265
  return -1;
dengyihao's avatar
dengyihao 已提交
1266
_return2:
dengyihao's avatar
dengyihao 已提交
1267
  tDebug("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1268
  return -1;
dengyihao's avatar
dengyihao 已提交
1269
}
1270

dengyihao's avatar
dengyihao 已提交
1271
int transSendResponse(const STransMsg* msg) {
dengyihao's avatar
dengyihao 已提交
1272 1273 1274 1275 1276
  if (msg->info.noResp) {
    rpcFreeCont(msg->pCont);
    tTrace("no need send resp");
    return 0;
  }
S
Shengliang Guan 已提交
1277 1278
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1279 1280
  ASYNC_CHECK_HANDLE(exh, refId);

dengyihao's avatar
dengyihao 已提交
1281
  STransMsg tmsg = *msg;
S
Shengliang Guan 已提交
1282
  tmsg.info.refId = refId;
dengyihao's avatar
dengyihao 已提交
1283

dengyihao's avatar
dengyihao 已提交
1284
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1285
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1286

dengyihao's avatar
dengyihao 已提交
1287 1288 1289
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Normal;
dengyihao's avatar
dengyihao 已提交
1290 1291

  STraceId* trace = (STraceId*)&msg->info.traceId;
dengyihao's avatar
dengyihao 已提交
1292
  tGDebug("conn %p start to send resp (1/2)", exh->handle);
1293 1294 1295
  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
    destroySmsg(m);
  }
H
Haojun Liao 已提交
1296

dengyihao's avatar
dengyihao 已提交
1297
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1298
  return 0;
H
Haojun Liao 已提交
1299

dengyihao's avatar
dengyihao 已提交
1300
_return1:
dengyihao's avatar
dengyihao 已提交
1301
  tDebug("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1302
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1303
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1304
  return -1;
dengyihao's avatar
dengyihao 已提交
1305
_return2:
dengyihao's avatar
dengyihao 已提交
1306
  tDebug("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1307
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1308
  return -1;
dengyihao's avatar
dengyihao 已提交
1309
}
dengyihao's avatar
dengyihao 已提交
1310
int transRegisterMsg(const STransMsg* msg) {
S
Shengliang Guan 已提交
1311 1312
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1313 1314
  ASYNC_CHECK_HANDLE(exh, refId);

dengyihao's avatar
dengyihao 已提交
1315
  STransMsg tmsg = *msg;
dengyihao's avatar
dengyihao 已提交
1316 1317
  tmsg.info.noResp = 1;

S
Shengliang Guan 已提交
1318
  tmsg.info.refId = refId;
dengyihao's avatar
dengyihao 已提交
1319

dengyihao's avatar
dengyihao 已提交
1320
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1321
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1322

dengyihao's avatar
dengyihao 已提交
1323 1324 1325
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Register;
dengyihao's avatar
dengyihao 已提交
1326

dengyihao's avatar
dengyihao 已提交
1327
  STrans* pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
1328
  tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
1329 1330 1331
  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
    destroySmsg(m);
  }
H
Haojun Liao 已提交
1332

dengyihao's avatar
dengyihao 已提交
1333
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1334
  return 0;
dengyihao's avatar
dengyihao 已提交
1335

dengyihao's avatar
dengyihao 已提交
1336
_return1:
dengyihao's avatar
dengyihao 已提交
1337
  tDebug("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1338
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1339
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1340
  return -1;
dengyihao's avatar
dengyihao 已提交
1341
_return2:
dengyihao's avatar
dengyihao 已提交
1342
  tDebug("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1343
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1344
  return -1;
dengyihao's avatar
dengyihao 已提交
1345
}
dengyihao's avatar
dengyihao 已提交
1346

D
dapan1121 已提交
1347
int transGetConnInfo(void* thandle, STransHandleInfo* pConnInfo) { return -1; }