transSvr.c 40.0 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
  if (cli == NULL) return;
dengyihao's avatar
dengyihao 已提交
681

dengyihao's avatar
dengyihao 已提交
682 683 684 685 686 687 688 689
  int err = uv_tcp_init(pObj->loop, cli);
  if (err != 0) {
    tError("failed to create tcp: %s", uv_err_name(err));
    taosMemoryFree(cli);
    return;
  }
  err = uv_accept(stream, (uv_stream_t*)cli);
  if (err == 0) {
dengyihao's avatar
dengyihao 已提交
690
#if defined(WINDOWS) || defined(DARWIN)
wafwerar's avatar
wafwerar 已提交
691
    if (pObj->numOfWorkerReady < pObj->numOfThreads) {
dengyihao's avatar
dengyihao 已提交
692 693
      tError("worker-threads are not ready for all, need %d instead of %d.", pObj->numOfThreads,
             pObj->numOfWorkerReady);
dengyihao's avatar
dengyihao 已提交
694
      uv_close((uv_handle_t*)cli, uvFreeCb);
wafwerar's avatar
wafwerar 已提交
695 696
      return;
    }
dengyihao's avatar
dengyihao 已提交
697
#endif
dengyihao's avatar
dengyihao 已提交
698

wafwerar's avatar
wafwerar 已提交
699
    uv_write_t* wr = (uv_write_t*)taosMemoryMalloc(sizeof(uv_write_t));
dengyihao's avatar
dengyihao 已提交
700
    wr->data = cli;
dengyihao's avatar
dengyihao 已提交
701 702 703
    uv_buf_t buf = uv_buf_init((char*)notify, strlen(notify));

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

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

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

dengyihao's avatar
dengyihao 已提交
735
  SWorkThrd* pThrd = q->data;
dengyihao's avatar
dengyihao 已提交
736 737 738 739 740 741 742 743 744

  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 已提交
745
  SSvrConn* pConn = createConn(pThrd);
746

dengyihao's avatar
dengyihao 已提交
747
  pConn->pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
748
  /* init conn timer*/
dengyihao's avatar
dengyihao 已提交
749 750
  // uv_timer_init(pThrd->loop, &pConn->pTimer);
  // pConn->pTimer.data = pConn;
dengyihao's avatar
dengyihao 已提交
751 752 753 754

  pConn->hostThrd = pThrd;

  // init client handle
wafwerar's avatar
wafwerar 已提交
755
  pConn->pTcp = (uv_tcp_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
756 757 758
  uv_tcp_init(pThrd->loop, pConn->pTcp);
  pConn->pTcp->data = pConn;

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

dengyihao's avatar
dengyihao 已提交
761 762 763
  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 已提交
764
    tTrace("conn %p created, fd:%d", pConn, fd);
dengyihao's avatar
dengyihao 已提交
765

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

dengyihao's avatar
dengyihao 已提交
775 776
    addrlen = sizeof(sockname);
    if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&sockname, &addrlen)) {
dengyihao's avatar
dengyihao 已提交
777
      tError("conn %p failed to get local info", pConn);
dengyihao's avatar
dengyihao 已提交
778
      transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
779 780
      return;
    }
dengyihao's avatar
dengyihao 已提交
781
    transSockInfo2Str(&sockname, pConn->src);
dengyihao's avatar
dengyihao 已提交
782

dengyihao's avatar
dengyihao 已提交
783
    struct sockaddr_in addr = *(struct sockaddr_in*)&peername;
dengyihao's avatar
dengyihao 已提交
784 785
    pConn->clientIp = addr.sin_addr.s_addr;
    pConn->port = ntohs(addr.sin_port);
dengyihao's avatar
dengyihao 已提交
786

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

dengyihao's avatar
dengyihao 已提交
789
  } else {
dengyihao's avatar
dengyihao 已提交
790
    tDebug("failed to create new connection");
dengyihao's avatar
dengyihao 已提交
791
    transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
792 793 794
  }
}

795
void* transAcceptThread(void* arg) {
dengyihao's avatar
dengyihao 已提交
796
  // opt
dengyihao's avatar
dengyihao 已提交
797
  setThreadName("trans-accept");
dengyihao's avatar
dengyihao 已提交
798 799
  SServerObj* srv = (SServerObj*)arg;
  uv_run(srv->loop, UV_RUN_DEFAULT);
800 801

  return NULL;
dengyihao's avatar
dengyihao 已提交
802
}
dengyihao's avatar
dengyihao 已提交
803
void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
wafwerar's avatar
wafwerar 已提交
804 805 806
  if (status != 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
807
  SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
wafwerar's avatar
wafwerar 已提交
808 809
  uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
}
dengyihao's avatar
dengyihao 已提交
810
static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
wafwerar's avatar
wafwerar 已提交
811
  pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
812 813 814
  if (0 != uv_loop_init(pThrd->loop)) {
    return false;
  }
dengyihao's avatar
dengyihao 已提交
815

dengyihao's avatar
dengyihao 已提交
816
#if defined(WINDOWS) || defined(DARWIN)
dengyihao's avatar
dengyihao 已提交
817
  uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
dengyihao's avatar
dengyihao 已提交
818 819 820 821
#else
  uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
  uv_pipe_open(pThrd->pipe, pThrd->fd);
#endif
dengyihao's avatar
dengyihao 已提交
822 823 824

  pThrd->pipe->data = pThrd;

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

dengyihao's avatar
dengyihao 已提交
827 828 829 830 831
  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 已提交
832 833 834
  // conn set
  QUEUE_INIT(&pThrd->conn);

dengyihao's avatar
dengyihao 已提交
835
  pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 8, pThrd, uvWorkerAsyncCb);
dengyihao's avatar
dengyihao 已提交
836
#if defined(WINDOWS) || defined(DARWIN)
wafwerar's avatar
wafwerar 已提交
837
  uv_pipe_connect(&pThrd->connect_req, pThrd->pipe, pipeName, uvOnPipeConnectionCb);
dengyihao's avatar
dengyihao 已提交
838 839 840
#else
  uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
#endif
dengyihao's avatar
dengyihao 已提交
841 842 843 844 845 846 847 848 849
  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 已提交
850
    tError("failed to init accept server:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
851 852 853
    return false;
  }

dengyihao's avatar
dengyihao 已提交
854
  // register an async here to quit server gracefully
wafwerar's avatar
wafwerar 已提交
855
  srv->pAcceptAsync = taosMemoryCalloc(1, sizeof(uv_async_t));
dengyihao's avatar
dengyihao 已提交
856 857
  uv_async_init(srv->loop, srv->pAcceptAsync, uvAcceptAsyncCb);
  srv->pAcceptAsync->data = srv;
dengyihao's avatar
dengyihao 已提交
858

dengyihao's avatar
dengyihao 已提交
859
  struct sockaddr_in bind_addr;
dengyihao's avatar
dengyihao 已提交
860 861
  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 已提交
862
    tError("failed to bind:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
863 864
    return false;
  }
dengyihao's avatar
dengyihao 已提交
865
  if ((err = uv_listen((uv_stream_t*)&srv->server, 4096 * 2, uvOnAcceptCb)) != 0) {
S
Shengliang Guan 已提交
866
    tError("failed to listen:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
867
    terrno = TSDB_CODE_RPC_PORT_EADDRINUSE;
dengyihao's avatar
dengyihao 已提交
868 869 870
    return false;
  }
  return true;
dengyihao's avatar
dengyihao 已提交
871
}
872
void* transWorkerThread(void* arg) {
G
gccgdb1234 已提交
873
  setThreadName("trans-svr-work");
dengyihao's avatar
dengyihao 已提交
874
  SWorkThrd* pThrd = (SWorkThrd*)arg;
dengyihao's avatar
dengyihao 已提交
875
  uv_run(pThrd->loop, UV_RUN_DEFAULT);
876 877

  return NULL;
dengyihao's avatar
dengyihao 已提交
878 879
}

dengyihao's avatar
dengyihao 已提交
880
static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
dengyihao's avatar
dengyihao 已提交
881
  SWorkThrd* pThrd = hThrd;
dengyihao's avatar
fix bug  
dengyihao 已提交
882

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

dengyihao's avatar
dengyihao 已提交
885
  transReqQueueInit(&pConn->wreqQueue);
dengyihao's avatar
fix bug  
dengyihao 已提交
886 887 888
  QUEUE_INIT(&pConn->queue);

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

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

dengyihao's avatar
dengyihao 已提交
892
  memset(&pConn->regArg, 0, sizeof(pConn->regArg));
dengyihao's avatar
dengyihao 已提交
893
  pConn->broken = false;
dengyihao's avatar
dengyihao 已提交
894
  pConn->status = ConnNormal;
dengyihao's avatar
opt rpc  
dengyihao 已提交
895
  transInitBuffer(&pConn->readBuf);
dengyihao's avatar
dengyihao 已提交
896

dengyihao's avatar
dengyihao 已提交
897 898 899
  SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
  exh->handle = pConn;
  exh->pThrd = pThrd;
dengyihao's avatar
dengyihao 已提交
900 901
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
902

dengyihao's avatar
dengyihao 已提交
903
  STrans* pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
904
  pConn->refId = exh->refId;
dengyihao's avatar
dengyihao 已提交
905
  transRefSrvHandle(pConn);
S
Shengliang Guan 已提交
906
  tTrace("%s handle %p, conn %p created, refId:%" PRId64, transLabel(pTransInst), exh, pConn, pConn->refId);
dengyihao's avatar
dengyihao 已提交
907 908
  return pConn;
}
dengyihao's avatar
dengyihao 已提交
909

dengyihao's avatar
dengyihao 已提交
910
static FORCE_INLINE void destroyConn(SSvrConn* conn, bool clear) {
dengyihao's avatar
dengyihao 已提交
911 912 913
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
914

915
  if (clear) {
dengyihao's avatar
dengyihao 已提交
916
    if (!uv_is_closing((uv_handle_t*)conn->pTcp)) {
dengyihao's avatar
dengyihao 已提交
917 918 919
      tTrace("conn %p to be destroyed", conn);
      uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn);
    }
920
  }
dengyihao's avatar
dengyihao 已提交
921
}
dengyihao's avatar
dengyihao 已提交
922
static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
923 924 925 926 927
  if (conn->regArg.init == 1) {
    transFreeMsg(conn->regArg.msg.pCont);
    conn->regArg.init = 0;
  }
}
dengyihao's avatar
dengyihao 已提交
928
static int reallocConnRef(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
929 930
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
931 932 933 934
  // 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 已提交
935 936
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
937 938 939 940
  conn->refId = exh->refId;

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

dengyihao's avatar
dengyihao 已提交
944 945 946
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
947
  SWorkThrd* thrd = conn->hostThrd;
dengyihao's avatar
dengyihao 已提交
948

dengyihao's avatar
dengyihao 已提交
949 950
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
951

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

dengyihao's avatar
dengyihao 已提交
955 956 957 958
  for (int i = 0; i < transQueueSize(&conn->srvMsgs); i++) {
    SSvrMsg* msg = transQueueGet(&conn->srvMsgs, i);
    destroySmsg(msg);
  }
dengyihao's avatar
dengyihao 已提交
959
  transQueueDestroy(&conn->srvMsgs);
dengyihao's avatar
dengyihao 已提交
960
  transReqQueueClear(&conn->wreqQueue);
dengyihao's avatar
dengyihao 已提交
961

dengyihao's avatar
dengyihao 已提交
962
  QUEUE_REMOVE(&conn->queue);
wafwerar's avatar
wafwerar 已提交
963
  taosMemoryFree(conn->pTcp);
dengyihao's avatar
dengyihao 已提交
964
  destroyConnRegArg(conn);
dengyihao's avatar
dengyihao 已提交
965
  transDestroyBuffer(&conn->readBuf);
dengyihao's avatar
dengyihao 已提交
966
  taosMemoryFree(conn);
dengyihao's avatar
dengyihao 已提交
967

dengyihao's avatar
dengyihao 已提交
968
  if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
dengyihao's avatar
dengyihao 已提交
969
    tTrace("work thread quit");
dengyihao's avatar
dengyihao 已提交
970
    uv_walk(thrd->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
971
  }
dengyihao's avatar
dengyihao 已提交
972
}
wafwerar's avatar
wafwerar 已提交
973
static void uvPipeListenCb(uv_stream_t* handle, int status) {
dengyihao's avatar
dengyihao 已提交
974
  if (status != 0) {
dengyihao's avatar
dengyihao 已提交
975
    tError("server failed to init pipe, errmsg: %s", uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
976 977
    return;
  }
wafwerar's avatar
wafwerar 已提交
978 979

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

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

dengyihao's avatar
dengyihao 已提交
986
  ret = uv_accept((uv_stream_t*)&srv->pipeListen, (uv_stream_t*)pipe);
dengyihao's avatar
dengyihao 已提交
987 988
  ASSERTS(ret == 0, "trans-svr failed to accept pipe msg");
  if (ret != 0) return;
dengyihao's avatar
dengyihao 已提交
989 990

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

dengyihao's avatar
dengyihao 已提交
994
  ret = uv_is_writable((uv_stream_t*)pipe);
dengyihao's avatar
dengyihao 已提交
995
  ASSERTS(ret == 1, "trans-svr pipe status corrupted");
dengyihao's avatar
dengyihao 已提交
996
  if (ret != 1) return;
wafwerar's avatar
wafwerar 已提交
997

dengyihao's avatar
dengyihao 已提交
998
  ret = uv_is_closing((uv_handle_t*)pipe);
dengyihao's avatar
dengyihao 已提交
999 1000
  ASSERTS(ret == 0, "trans-svr pipe status corrupted");
  if (ret != 0) return;
dengyihao's avatar
dengyihao 已提交
1001 1002

  srv->numOfWorkerReady++;
wafwerar's avatar
wafwerar 已提交
1003
}
dengyihao's avatar
dengyihao 已提交
1004

U
ubuntu 已提交
1005
void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
wafwerar's avatar
wafwerar 已提交
1006 1007
  SServerObj* srv = taosMemoryCalloc(1, sizeof(SServerObj));
  srv->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
1008 1009
  srv->numOfThreads = numOfThreads;
  srv->workerIdx = 0;
wafwerar's avatar
wafwerar 已提交
1010
  srv->numOfWorkerReady = 0;
dengyihao's avatar
dengyihao 已提交
1011
  srv->pThreadObj = (SWorkThrd**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrd*));
wafwerar's avatar
wafwerar 已提交
1012
  srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*));
dengyihao's avatar
dengyihao 已提交
1013 1014 1015 1016
  srv->ip = ip;
  srv->port = port;
  uv_loop_init(srv->loop);

dengyihao's avatar
dengyihao 已提交
1017
  char pipeName[PATH_MAX];
1018 1019 1020 1021 1022 1023 1024

  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 已提交
1025
#if defined(WINDOWS) || defined(DARWIN)
dengyihao's avatar
dengyihao 已提交
1026
  int ret = uv_pipe_init(srv->loop, &srv->pipeListen, 0);
dengyihao's avatar
dengyihao 已提交
1027 1028 1029 1030
  if (ret != 0) {
    tError("failed to init pipe, errmsg: %s", uv_err_name(ret));
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1031
#if defined(WINDOWS)
dengyihao's avatar
dengyihao 已提交
1032
  snprintf(pipeName, sizeof(pipeName), "\\\\?\\pipe\\trans.rpc.%d-%" PRIu64, taosSafeRand(), GetCurrentProcessId());
dengyihao's avatar
dengyihao 已提交
1033 1034 1035 1036
#elif defined(DARWIN)
  snprintf(pipeName, sizeof(pipeName), "%s%spipe.trans.rpc.%08d-%" PRIu64, tsTempDir, TD_DIRSEP, taosSafeRand(),
           taosGetSelfPthreadId());
#endif
dengyihao's avatar
dengyihao 已提交
1037

dengyihao's avatar
dengyihao 已提交
1038
  ret = uv_pipe_bind(&srv->pipeListen, pipeName);
dengyihao's avatar
dengyihao 已提交
1039 1040 1041 1042
  if (ret != 0) {
    tError("failed to bind pipe, errmsg: %s", uv_err_name(ret));
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1043 1044

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

dengyihao's avatar
dengyihao 已提交
1050
  for (int i = 0; i < srv->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
1051
    SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));
dengyihao's avatar
dengyihao 已提交
1052
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
1053
    thrd->quit = false;
dengyihao's avatar
dengyihao 已提交
1054
    srv->pThreadObj[i] = thrd;
dengyihao's avatar
dengyihao 已提交
1055
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
1056

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

dengyihao's avatar
dengyihao 已提交
1060
    if (false == addHandleToWorkloop(thrd, pipeName)) {
dengyihao's avatar
dengyihao 已提交
1061 1062
      goto End;
    }
dengyihao's avatar
dengyihao 已提交
1063

dengyihao's avatar
dengyihao 已提交
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085
    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 已提交
1086
    if (uv_socketpair(SOCK_STREAM, 0, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) {
dengyihao's avatar
dengyihao 已提交
1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099
      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;
    }

1100
    int err = taosThreadCreate(&(thrd->thread), NULL, transWorkerThread, (void*)(thrd));
dengyihao's avatar
dengyihao 已提交
1101
    if (err == 0) {
S
Shengliang Guan 已提交
1102
      tDebug("success to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
1103 1104
    } else {
      // TODO: clear all other resource later
S
Shengliang Guan 已提交
1105
      tError("failed to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
1106
      goto End;
dengyihao's avatar
dengyihao 已提交
1107 1108
    }
  }
dengyihao's avatar
dengyihao 已提交
1109
#endif
dengyihao's avatar
dengyihao 已提交
1110

dengyihao's avatar
dengyihao 已提交
1111 1112 1113
  if (false == addHandleToAcceptloop(srv)) {
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
1114

1115
  int err = taosThreadCreate(&srv->thread, NULL, transAcceptThread, (void*)srv);
dengyihao's avatar
dengyihao 已提交
1116 1117 1118
  if (err == 0) {
    tDebug("success to create accept-thread");
  } else {
dengyihao's avatar
dengyihao 已提交
1119 1120
    tError("failed  to create accept-thread");
    goto End;
dengyihao's avatar
dengyihao 已提交
1121 1122
    // clear all resource later
  }
dengyihao's avatar
dengyihao 已提交
1123

dengyihao's avatar
dengyihao 已提交
1124
  srv->inited = true;
dengyihao's avatar
dengyihao 已提交
1125
  return srv;
dengyihao's avatar
dengyihao 已提交
1126
End:
U
ubuntu 已提交
1127
  transCloseServer(srv);
dengyihao's avatar
dengyihao 已提交
1128 1129
  return NULL;
}
dengyihao's avatar
dengyihao 已提交
1130

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

    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 已提交
1177
    taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
1178 1179
  }
}
dengyihao's avatar
dengyihao 已提交
1180
void destroyWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
1181 1182 1183
  if (pThrd == NULL) {
    return;
  }
wafwerar's avatar
wafwerar 已提交
1184
  taosThreadJoin(pThrd->thread, NULL);
dengyihao's avatar
dengyihao 已提交
1185
  SRV_RELEASE_UV(pThrd->loop);
dengyihao's avatar
dengyihao 已提交
1186
  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg);
dengyihao's avatar
dengyihao 已提交
1187
  transAsyncPoolDestroy(pThrd->asyncPool);
dengyihao's avatar
dengyihao 已提交
1188
  taosMemoryFree(pThrd->prepare);
dengyihao's avatar
dengyihao 已提交
1189
  taosMemoryFree(pThrd->loop);
wafwerar's avatar
wafwerar 已提交
1190
  taosMemoryFree(pThrd);
dengyihao's avatar
dengyihao 已提交
1191
}
dengyihao's avatar
dengyihao 已提交
1192
void sendQuitToWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
1193
  SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
dengyihao's avatar
dengyihao 已提交
1194
  msg->type = Quit;
dengyihao's avatar
dengyihao 已提交
1195
  tDebug("server send quit msg to work thread");
dengyihao's avatar
dengyihao 已提交
1196
  transAsyncSend(pThrd->asyncPool, &msg->q);
dengyihao's avatar
dengyihao 已提交
1197 1198
}

U
ubuntu 已提交
1199
void transCloseServer(void* arg) {
dengyihao's avatar
dengyihao 已提交
1200 1201
  // impl later
  SServerObj* srv = arg;
dengyihao's avatar
dengyihao 已提交
1202

dengyihao's avatar
dengyihao 已提交
1203
  if (srv->inited) {
1204
    tDebug("send quit msg to accept thread");
dengyihao's avatar
dengyihao 已提交
1205 1206
    uv_async_send(srv->pAcceptAsync);
    taosThreadJoin(srv->thread, NULL);
dengyihao's avatar
dengyihao 已提交
1207
    SRV_RELEASE_UV(srv->loop);
dengyihao's avatar
dengyihao 已提交
1208

dengyihao's avatar
dengyihao 已提交
1209 1210 1211 1212
    for (int i = 0; i < srv->numOfThreads; i++) {
      sendQuitToWorkThrd(srv->pThreadObj[i]);
      destroyWorkThrd(srv->pThreadObj[i]);
    }
dengyihao's avatar
dengyihao 已提交
1213 1214
  } else {
    uv_loop_close(srv->loop);
dengyihao's avatar
dengyihao 已提交
1215 1216
  }

wafwerar's avatar
wafwerar 已提交
1217 1218 1219
  taosMemoryFree(srv->pThreadObj);
  taosMemoryFree(srv->pAcceptAsync);
  taosMemoryFree(srv->loop);
dengyihao's avatar
dengyihao 已提交
1220 1221

  for (int i = 0; i < srv->numOfThreads; i++) {
wafwerar's avatar
wafwerar 已提交
1222
    taosMemoryFree(srv->pipe[i]);
dengyihao's avatar
dengyihao 已提交
1223
  }
wafwerar's avatar
wafwerar 已提交
1224
  taosMemoryFree(srv->pipe);
dengyihao's avatar
dengyihao 已提交
1225

wafwerar's avatar
wafwerar 已提交
1226
  taosMemoryFree(srv);
dengyihao's avatar
dengyihao 已提交
1227
}
dengyihao's avatar
dengyihao 已提交
1228

dengyihao's avatar
dengyihao 已提交
1229 1230 1231 1232
void transRefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1233
  int ref = T_REF_INC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1234
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1235 1236 1237 1238 1239 1240
}

void transUnrefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1241
  int ref = T_REF_DEC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1242
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1243
  if (ref == 0) {
dengyihao's avatar
dengyihao 已提交
1244
    destroyConn((SSvrConn*)handle, true);
dengyihao's avatar
dengyihao 已提交
1245 1246
  }
}
dengyihao's avatar
dengyihao 已提交
1247

dengyihao's avatar
dengyihao 已提交
1248
int transReleaseSrvHandle(void* handle) {
dengyihao's avatar
dengyihao 已提交
1249 1250 1251
  SRpcHandleInfo* info = handle;
  SExHandle*      exh = info->handle;
  int64_t         refId = info->refId;
dengyihao's avatar
dengyihao 已提交
1252

dengyihao's avatar
dengyihao 已提交
1253
  ASYNC_CHECK_HANDLE(exh, refId);
dengyihao's avatar
dengyihao 已提交
1254

dengyihao's avatar
dengyihao 已提交
1255
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1256 1257
  ASYNC_ERR_JRET(pThrd);

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

dengyihao's avatar
dengyihao 已提交
1260 1261 1262
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Release;
dengyihao's avatar
dengyihao 已提交
1263

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

dengyihao's avatar
dengyihao 已提交
1269
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1270
  return 0;
dengyihao's avatar
dengyihao 已提交
1271
_return1:
dengyihao's avatar
dengyihao 已提交
1272
  tDebug("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1273
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1274
  return -1;
dengyihao's avatar
dengyihao 已提交
1275
_return2:
dengyihao's avatar
dengyihao 已提交
1276
  tDebug("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1277
  return -1;
dengyihao's avatar
dengyihao 已提交
1278
}
1279

dengyihao's avatar
dengyihao 已提交
1280
int transSendResponse(const STransMsg* msg) {
dengyihao's avatar
dengyihao 已提交
1281 1282 1283 1284 1285
  if (msg->info.noResp) {
    rpcFreeCont(msg->pCont);
    tTrace("no need send resp");
    return 0;
  }
S
Shengliang Guan 已提交
1286 1287
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1288 1289
  ASYNC_CHECK_HANDLE(exh, refId);

dengyihao's avatar
dengyihao 已提交
1290
  STransMsg tmsg = *msg;
S
Shengliang Guan 已提交
1291
  tmsg.info.refId = refId;
dengyihao's avatar
dengyihao 已提交
1292

dengyihao's avatar
dengyihao 已提交
1293
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1294
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1295

dengyihao's avatar
dengyihao 已提交
1296 1297 1298
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Normal;
dengyihao's avatar
dengyihao 已提交
1299 1300

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

dengyihao's avatar
dengyihao 已提交
1306
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1307
  return 0;
H
Haojun Liao 已提交
1308

dengyihao's avatar
dengyihao 已提交
1309
_return1:
dengyihao's avatar
dengyihao 已提交
1310
  tDebug("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1311
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1312
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1313
  return -1;
dengyihao's avatar
dengyihao 已提交
1314
_return2:
dengyihao's avatar
dengyihao 已提交
1315
  tDebug("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1316
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1317
  return -1;
dengyihao's avatar
dengyihao 已提交
1318
}
dengyihao's avatar
dengyihao 已提交
1319
int transRegisterMsg(const STransMsg* msg) {
S
Shengliang Guan 已提交
1320 1321
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1322 1323
  ASYNC_CHECK_HANDLE(exh, refId);

dengyihao's avatar
dengyihao 已提交
1324
  STransMsg tmsg = *msg;
dengyihao's avatar
dengyihao 已提交
1325 1326
  tmsg.info.noResp = 1;

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

dengyihao's avatar
dengyihao 已提交
1329
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1330
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1331

dengyihao's avatar
dengyihao 已提交
1332 1333 1334
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Register;
dengyihao's avatar
dengyihao 已提交
1335

dengyihao's avatar
dengyihao 已提交
1336
  STrans* pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
1337
  tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
1338 1339 1340
  if (0 != transAsyncSend(pThrd->asyncPool, &m->q)) {
    destroySmsg(m);
  }
H
Haojun Liao 已提交
1341

dengyihao's avatar
dengyihao 已提交
1342
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1343
  return 0;
dengyihao's avatar
dengyihao 已提交
1344

dengyihao's avatar
dengyihao 已提交
1345
_return1:
dengyihao's avatar
dengyihao 已提交
1346
  tDebug("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1347
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1348
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1349
  return -1;
dengyihao's avatar
dengyihao 已提交
1350
_return2:
dengyihao's avatar
dengyihao 已提交
1351
  tDebug("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1352
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1353
  return -1;
dengyihao's avatar
dengyihao 已提交
1354
}
dengyihao's avatar
dengyihao 已提交
1355

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