transSvr.c 35.2 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 15 16
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#ifdef USE_UV

17
#include "transComm.h"
dengyihao's avatar
dengyihao 已提交
18

dengyihao's avatar
dengyihao 已提交
19 20
static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;

dengyihao's avatar
dengyihao 已提交
21
static char* notify = "a";
dengyihao's avatar
dengyihao 已提交
22

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

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

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

dengyihao's avatar
dengyihao 已提交
43
  SSvrRegArg regArg;
dengyihao's avatar
dengyihao 已提交
44
  bool       broken;  // conn broken;
dengyihao's avatar
dengyihao 已提交
45

dengyihao's avatar
dengyihao 已提交
46
  ConnStatus         status;
dengyihao's avatar
dengyihao 已提交
47
  struct sockaddr_in addr;
dengyihao's avatar
dengyihao 已提交
48
  struct sockaddr_in localAddr;
U
ubuntu 已提交
49

dengyihao's avatar
dengyihao 已提交
50 51 52 53 54 55
  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 已提交
56
} SSvrConn;
dengyihao's avatar
dengyihao 已提交
57

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

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

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

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

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

wafwerar's avatar
wafwerar 已提交
91
  uv_pipe_t   pipeListen;
dengyihao's avatar
dengyihao 已提交
92 93 94 95
  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 已提交
96 97

  bool inited;
dengyihao's avatar
dengyihao 已提交
98 99 100
} SServerObj;

static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
101
static void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
102
static void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf);
dengyihao's avatar
dengyihao 已提交
103
static void uvOnTimeoutCb(uv_timer_t* handle);
dengyihao's avatar
dengyihao 已提交
104
static void uvOnSendCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
105
static void uvOnPipeWriteCb(uv_write_t* req, int status);
dengyihao's avatar
dengyihao 已提交
106 107 108
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 已提交
109
static void uvAcceptAsyncCb(uv_async_t* handle);
dengyihao's avatar
dengyihao 已提交
110
static void uvShutDownCb(uv_shutdown_t* req, int status);
dengyihao's avatar
dengyihao 已提交
111 112 113 114 115 116 117

/*
 * 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 已提交
118 119
static void uvWalkCb(uv_handle_t* handle, void* arg);
static void uvFreeCb(uv_handle_t* handle);
dengyihao's avatar
dengyihao 已提交
120

dengyihao's avatar
dengyihao 已提交
121 122 123
static void uvStartSendRespInternal(SSvrMsg* smsg);
static void uvPrepareSendData(SSvrMsg* msg, uv_buf_t* wb);
static void uvStartSendResp(SSvrMsg* msg);
dengyihao's avatar
dengyihao 已提交
124

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

dengyihao's avatar
dengyihao 已提交
127
static void destroySmsg(SSvrMsg* smsg);
128
// check whether already read complete packet
dengyihao's avatar
dengyihao 已提交
129 130 131
static SSvrConn* createConn(void* hThrd);
static void      destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/);
static void      destroyConnRegArg(SSvrConn* conn);
dengyihao's avatar
dengyihao 已提交
132

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

dengyihao's avatar
dengyihao 已提交
135 136 137 138 139 140
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 已提交
141

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

dengyihao's avatar
dengyihao 已提交
144
// server and worker thread
145 146
static void* transWorkerThread(void* arg);
static void* transAcceptThread(void* arg);
dengyihao's avatar
dengyihao 已提交
147

dengyihao's avatar
dengyihao 已提交
148
// add handle loop
dengyihao's avatar
dengyihao 已提交
149
static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
dengyihao's avatar
dengyihao 已提交
150 151
static bool addHandleToAcceptloop(void* arg);

S
Shengliang Guan 已提交
152 153 154 155 156 157
#define CONN_SHOULD_RELEASE(conn, head)                                               \
  do {                                                                                \
    if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) {                    \
      conn->status = ConnRelease;                                                     \
      transClearBuffer(&conn->readBuf);                                               \
      transFreeMsg(transContFromHead((char*)head));                                   \
dengyihao's avatar
dengyihao 已提交
158
      tTrace("conn %p received release request", conn);                               \
S
Shengliang Guan 已提交
159 160
                                                                                      \
      STransMsg tmsg = {.code = 0, .info.handle = (void*)conn, .info.ahandle = NULL}; \
dengyihao's avatar
dengyihao 已提交
161
      SSvrMsg*  srvMsg = taosMemoryCalloc(1, sizeof(SSvrMsg));                        \
S
Shengliang Guan 已提交
162 163 164
      srvMsg->msg = tmsg;                                                             \
      srvMsg->type = Release;                                                         \
      srvMsg->pConn = conn;                                                           \
dengyihao's avatar
dengyihao 已提交
165
      reallocConnRef(conn);                                                           \
S
Shengliang Guan 已提交
166 167 168
      if (!transQueuePush(&conn->srvMsgs, srvMsg)) {                                  \
        return;                                                                       \
      }                                                                               \
dengyihao's avatar
dengyihao 已提交
169
      if (conn->regArg.init) {                                                        \
dengyihao's avatar
dengyihao 已提交
170
        tTrace("conn %p release, notify server app", conn);                           \
dengyihao's avatar
dengyihao 已提交
171 172 173 174
        STrans* pTransInst = conn->pTransInst;                                        \
        (*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);            \
        memset(&conn->regArg, 0, sizeof(conn->regArg));                               \
      }                                                                               \
S
Shengliang Guan 已提交
175 176 177
      uvStartSendRespInternal(srvMsg);                                                \
      return;                                                                         \
    }                                                                                 \
dengyihao's avatar
dengyihao 已提交
178 179 180 181 182 183 184 185 186
  } while (0)

#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 已提交
187 188 189 190
#define ASYNC_ERR_JRET(thrd)                            \
  do {                                                  \
    if (thrd->quit) {                                   \
      tTrace("worker thread already quit, ignore msg"); \
dengyihao's avatar
dengyihao 已提交
191
      goto _return1;                                    \
dengyihao's avatar
dengyihao 已提交
192 193 194
    }                                                   \
  } while (0)

dengyihao's avatar
dengyihao 已提交
195
void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
196
  SSvrConn*    conn = handle->data;
dengyihao's avatar
dengyihao 已提交
197 198
  SConnBuffer* pBuf = &conn->readBuf;
  transAllocBuffer(pBuf, buf);
dengyihao's avatar
dengyihao 已提交
199 200 201 202
}

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

dengyihao's avatar
dengyihao 已提交
207
static void uvHandleReq(SSvrConn* pConn) {
dengyihao's avatar
dengyihao 已提交
208
  SConnBuffer* pBuf = &pConn->readBuf;
dengyihao's avatar
dengyihao 已提交
209 210 211 212
  char*        msg = pBuf->buf;
  uint32_t     msgLen = pBuf->len;

  STransMsgHead* pHead = (STransMsgHead*)msg;
dengyihao's avatar
dengyihao 已提交
213
  pHead->code = htonl(pHead->code);
dengyihao's avatar
dengyihao 已提交
214
  pHead->msgLen = htonl(pHead->msgLen);
dengyihao's avatar
dengyihao 已提交
215
  memcpy(pConn->user, pHead->user, strlen(pHead->user));
dengyihao's avatar
dengyihao 已提交
216

dengyihao's avatar
dengyihao 已提交
217 218 219 220 221
  // 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 已提交
222
  //  uv_queue_work(((SWorkThrd*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask);
dengyihao's avatar
dengyihao 已提交
223

dengyihao's avatar
dengyihao 已提交
224
  CONN_SHOULD_RELEASE(pConn, pHead);
dengyihao's avatar
dengyihao 已提交
225

dengyihao's avatar
dengyihao 已提交
226
  STransMsg transMsg;
dengyihao's avatar
dengyihao 已提交
227
  memset(&transMsg, 0, sizeof(transMsg));
dengyihao's avatar
dengyihao 已提交
228 229 230 231
  transMsg.contLen = transContLenFromMsg(pHead->msgLen);
  transMsg.pCont = pHead->content;
  transMsg.msgType = pHead->msgType;
  transMsg.code = pHead->code;
dengyihao's avatar
dengyihao 已提交
232

dengyihao's avatar
dengyihao 已提交
233
  transClearBuffer(&pConn->readBuf);
dengyihao's avatar
dengyihao 已提交
234

dengyihao's avatar
dengyihao 已提交
235
  pConn->inType = pHead->msgType;
dengyihao's avatar
dengyihao 已提交
236 237 238 239
  if (pConn->status == ConnNormal) {
    if (pHead->persist == 1) {
      pConn->status = ConnAcquire;
      transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
240
      tDebug("conn %p acquired by server app", pConn);
dengyihao's avatar
dengyihao 已提交
241 242
    }
  }
dengyihao's avatar
dengyihao 已提交
243
  STrans*   pTransInst = pConn->pTransInst;
dengyihao's avatar
dengyihao 已提交
244
  STraceId* trace = &pHead->traceId;
dengyihao's avatar
dengyihao 已提交
245
  if (pConn->status == ConnNormal && pHead->noResp == 0) {
dengyihao's avatar
dengyihao 已提交
246
    transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
247

S
Shengliang Guan 已提交
248
    tGTrace("%s conn %p %s received from %s:%d, local info:%s:%d, msg size:%d", transLabel(pTransInst), pConn,
dengyihao's avatar
dengyihao 已提交
249
            TMSG_INFO(transMsg.msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
dengyihao's avatar
dengyihao 已提交
250 251
            taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port), transMsg.contLen);
  } else {
S
Shengliang Guan 已提交
252
    tGTrace("%s conn %p %s received from %s:%d, local info:%s:%d, msg size:%d, resp:%d, code:%d",
dengyihao's avatar
dengyihao 已提交
253 254 255
            transLabel(pTransInst), pConn, TMSG_INFO(transMsg.msgType), taosInetNtoa(pConn->addr.sin_addr),
            ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port),
            transMsg.contLen, pHead->noResp, transMsg.code);
dengyihao's avatar
dengyihao 已提交
256 257 258
    // no ref here
  }

dengyihao's avatar
dengyihao 已提交
259
  // pHead->noResp = 1,
dengyihao's avatar
dengyihao 已提交
260 261 262
  // 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 已提交
263
  transMsg.info.ahandle = (void*)pHead->ahandle;
dengyihao's avatar
dengyihao 已提交
264
  transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId);
S
Shengliang Guan 已提交
265
  transMsg.info.refId = pConn->refId;
dengyihao's avatar
dengyihao 已提交
266 267
  transMsg.info.traceId = pHead->traceId;

dengyihao's avatar
dengyihao 已提交
268 269
  tGTrace("%s handle %p conn:%p translated to app, refId:%" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn,
          pConn->refId);
S
Shengliang Guan 已提交
270
  assert(transMsg.info.handle != NULL);
dengyihao's avatar
dengyihao 已提交
271

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

  // set up conn info
277
  SRpcConnInfo* pConnInfo = &(transMsg.info.conn);
dengyihao's avatar
dengyihao 已提交
278 279 280 281
  pConnInfo->clientIp = (uint32_t)(pConn->addr.sin_addr.s_addr);
  pConnInfo->clientPort = ntohs(pConn->addr.sin_port);
  tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user));

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

dengyihao's avatar
dengyihao 已提交
284
  (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
dengyihao's avatar
dengyihao 已提交
285
  // uv_timer_start(&pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
dengyihao's avatar
dengyihao 已提交
286 287
}

dengyihao's avatar
dengyihao 已提交
288
void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
289
  // opt
dengyihao's avatar
dengyihao 已提交
290
  SSvrConn*    conn = cli->data;
dengyihao's avatar
dengyihao 已提交
291
  SConnBuffer* pBuf = &conn->readBuf;
dengyihao's avatar
dengyihao 已提交
292
  STrans*      pTransInst = conn->pTransInst;
dengyihao's avatar
dengyihao 已提交
293 294
  if (nread > 0) {
    pBuf->len += nread;
S
Shengliang Guan 已提交
295
    tTrace("%s conn %p total read:%d, current read:%d", transLabel(pTransInst), conn, pBuf->len, (int)nread);
dengyihao's avatar
dengyihao 已提交
296
    if (transReadComplete(pBuf)) {
dengyihao's avatar
dengyihao 已提交
297
      tTrace("%s conn %p alread read complete packet", transLabel(pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
298
      uvHandleReq(conn);
dengyihao's avatar
dengyihao 已提交
299
    } else {
dengyihao's avatar
dengyihao 已提交
300
      tTrace("%s conn %p read partial packet, continue to read", transLabel(pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
301 302 303
    }
    return;
  }
304 305 306
  if (nread == 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
307

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

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

dengyihao's avatar
dengyihao 已提交
333
void uvOnSendCb(uv_write_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
334
  SSvrConn* conn = transReqQueueRemove(req);
dengyihao's avatar
dengyihao 已提交
335 336
  if (conn == NULL) return;

dengyihao's avatar
dengyihao 已提交
337
  if (status == 0) {
dengyihao's avatar
dengyihao 已提交
338
    tTrace("conn %p data already was written on stream", conn);
dengyihao's avatar
dengyihao 已提交
339
    if (!transQueueEmpty(&conn->srvMsgs)) {
dengyihao's avatar
dengyihao 已提交
340
      SSvrMsg* msg = transQueuePop(&conn->srvMsgs);
dengyihao's avatar
add UT  
dengyihao 已提交
341
      destroySmsg(msg);
dengyihao's avatar
dengyihao 已提交
342
      // send cached data
dengyihao's avatar
dengyihao 已提交
343
      if (!transQueueEmpty(&conn->srvMsgs)) {
dengyihao's avatar
dengyihao 已提交
344
        msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
dengyihao's avatar
dengyihao 已提交
345 346 347 348 349 350 351 352 353
        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 已提交
354
          transQueuePop(&conn->srvMsgs);
wafwerar's avatar
wafwerar 已提交
355
          taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
356

dengyihao's avatar
dengyihao 已提交
357
          msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
dengyihao's avatar
dengyihao 已提交
358 359 360
          if (msg != NULL) {
            uvStartSendRespInternal(msg);
          }
dengyihao's avatar
dengyihao 已提交
361 362 363
        } else {
          uvStartSendRespInternal(msg);
        }
dengyihao's avatar
add UT  
dengyihao 已提交
364
      }
dengyihao's avatar
dengyihao 已提交
365
    }
dengyihao's avatar
dengyihao 已提交
366
    transUnrefSrvHandle(conn);
dengyihao's avatar
dengyihao 已提交
367
  } else {
dengyihao's avatar
dengyihao 已提交
368
    tError("conn %p failed to write data, %s", conn, uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
369
    conn->broken = true;
dengyihao's avatar
dengyihao 已提交
370
    transUnrefSrvHandle(conn);
dengyihao's avatar
dengyihao 已提交
371 372
  }
}
dengyihao's avatar
dengyihao 已提交
373 374
static void uvOnPipeWriteCb(uv_write_t* req, int status) {
  if (status == 0) {
dengyihao's avatar
dengyihao 已提交
375
    tTrace("success to dispatch conn to work thread");
dengyihao's avatar
dengyihao 已提交
376 377 378
  } else {
    tError("fail to dispatch conn to work thread");
  }
dengyihao's avatar
dengyihao 已提交
379
  uv_close((uv_handle_t*)req->data, uvFreeCb);
wafwerar's avatar
wafwerar 已提交
380
  taosMemoryFree(req);
dengyihao's avatar
dengyihao 已提交
381
}
dengyihao's avatar
dengyihao 已提交
382

dengyihao's avatar
dengyihao 已提交
383 384
static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
  SSvrConn*  pConn = smsg->pConn;
dengyihao's avatar
formate  
dengyihao 已提交
385
  STransMsg* pMsg = &smsg->msg;
dengyihao's avatar
dengyihao 已提交
386 387 388 389 390
  if (pMsg->pCont == 0) {
    pMsg->pCont = (void*)rpcMallocCont(0);
    pMsg->contLen = 0;
  }
  STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
S
Shengliang Guan 已提交
391
  pHead->ahandle = (uint64_t)pMsg->info.ahandle;
dengyihao's avatar
dengyihao 已提交
392
  pHead->traceId = pMsg->info.traceId;
dengyihao's avatar
dengyihao 已提交
393
  pHead->hasEpSet = pMsg->info.hasEpSet;
dengyihao's avatar
dengyihao 已提交
394

dengyihao's avatar
dengyihao 已提交
395
  if (pConn->status == ConnNormal) {
dengyihao's avatar
dengyihao 已提交
396
    pHead->msgType = (0 == pMsg->msgType ? pConn->inType + 1 : pMsg->msgType);
dengyihao's avatar
dengyihao 已提交
397
  } else {
dengyihao's avatar
dengyihao 已提交
398 399 400
    if (smsg->type == Release) {
      pHead->msgType = 0;
      pConn->status = ConnNormal;
dengyihao's avatar
dengyihao 已提交
401 402

      destroyConnRegArg(pConn);
dengyihao's avatar
dengyihao 已提交
403 404
      transUnrefSrvHandle(pConn);
    } else {
dengyihao's avatar
dengyihao 已提交
405
      // set up resp msg type
dengyihao's avatar
dengyihao 已提交
406
      pHead->msgType = (0 == pMsg->msgType ? pConn->inType + 1 : pMsg->msgType);
dengyihao's avatar
dengyihao 已提交
407
    }
dengyihao's avatar
dengyihao 已提交
408
  }
dengyihao's avatar
dengyihao 已提交
409

dengyihao's avatar
dengyihao 已提交
410
  pHead->release = smsg->type == Release ? 1 : 0;
dengyihao's avatar
dengyihao 已提交
411
  pHead->code = htonl(pMsg->code);
dengyihao's avatar
dengyihao 已提交
412

dengyihao's avatar
dengyihao 已提交
413 414
  char*   msg = (char*)pHead;
  int32_t len = transMsgLenFromCont(pMsg->contLen);
dengyihao's avatar
dengyihao 已提交
415

dengyihao's avatar
dengyihao 已提交
416
  STrans*   pTransInst = pConn->pTransInst;
dengyihao's avatar
dengyihao 已提交
417
  STraceId* trace = &pMsg->info.traceId;
S
Shengliang Guan 已提交
418
  tGTrace("%s conn %p %s is sent to %s:%d, local info:%s:%d, msglen:%d", transLabel(pTransInst), pConn,
dengyihao's avatar
dengyihao 已提交
419 420
          TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
          taosInetNtoa(pConn->localAddr.sin_addr), ntohs(pConn->localAddr.sin_port), len);
dengyihao's avatar
dengyihao 已提交
421
  pHead->msgLen = htonl(len);
dengyihao's avatar
dengyihao 已提交
422

dengyihao's avatar
dengyihao 已提交
423 424 425
  wb->base = msg;
  wb->len = len;
}
dengyihao's avatar
dengyihao 已提交
426

dengyihao's avatar
dengyihao 已提交
427
static void uvStartSendRespInternal(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
428 429 430 431 432
  SSvrConn* pConn = smsg->pConn;
  if (pConn->broken) {
    return;
  }

dengyihao's avatar
dengyihao 已提交
433 434 435
  uv_buf_t wb;
  uvPrepareSendData(smsg, &wb);

dengyihao's avatar
dengyihao 已提交
436
  transRefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
437
  uv_write_t* req = transReqQueuePushReq(&pConn->wreqQueue);
dengyihao's avatar
dengyihao 已提交
438
  uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb);
dengyihao's avatar
dengyihao 已提交
439
}
dengyihao's avatar
dengyihao 已提交
440
static void uvStartSendResp(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
441
  // impl
dengyihao's avatar
dengyihao 已提交
442
  SSvrConn* pConn = smsg->pConn;
dengyihao's avatar
dengyihao 已提交
443
  if (pConn->broken == true) {
dengyihao's avatar
dengyihao 已提交
444
    // persist by
dengyihao's avatar
dengyihao 已提交
445 446
    transFreeMsg(smsg->msg.pCont);
    taosMemoryFree(smsg);
dengyihao's avatar
dengyihao 已提交
447 448 449
    transUnrefSrvHandle(pConn);
    return;
  }
dengyihao's avatar
dengyihao 已提交
450 451 452
  if (pConn->status == ConnNormal) {
    transUnrefSrvHandle(pConn);
  }
dengyihao's avatar
dengyihao 已提交
453

dengyihao's avatar
dengyihao 已提交
454
  if (!transQueuePush(&pConn->srvMsgs, smsg)) {
dengyihao's avatar
dengyihao 已提交
455 456 457
    return;
  }
  uvStartSendRespInternal(smsg);
dengyihao's avatar
dengyihao 已提交
458 459
  return;
}
dengyihao's avatar
dengyihao 已提交
460

dengyihao's avatar
dengyihao 已提交
461
static void destroySmsg(SSvrMsg* smsg) {
dengyihao's avatar
dengyihao 已提交
462 463 464 465
  if (smsg == NULL) {
    return;
  }
  transFreeMsg(smsg->msg.pCont);
wafwerar's avatar
wafwerar 已提交
466
  taosMemoryFree(smsg);
dengyihao's avatar
dengyihao 已提交
467
}
dengyihao's avatar
dengyihao 已提交
468
static void destroyAllConn(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
469
  tTrace("thread %p destroy all conn ", pThrd);
dengyihao's avatar
fix bug  
dengyihao 已提交
470 471 472 473 474
  while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
    queue* h = QUEUE_HEAD(&pThrd->conn);
    QUEUE_REMOVE(h);
    QUEUE_INIT(h);

dengyihao's avatar
dengyihao 已提交
475
    SSvrConn* c = QUEUE_DATA(h, SSvrConn, queue);
dengyihao's avatar
dengyihao 已提交
476 477 478
    while (T_REF_VAL_GET(c) >= 2) {
      transUnrefSrvHandle(c);
    }
dengyihao's avatar
dengyihao 已提交
479
    transUnrefSrvHandle(c);
dengyihao's avatar
fix bug  
dengyihao 已提交
480 481
  }
}
dengyihao's avatar
dengyihao 已提交
482
void uvWorkerAsyncCb(uv_async_t* handle) {
dengyihao's avatar
dengyihao 已提交
483 484 485 486
  SAsyncItem* item = handle->data;
  SWorkThrd*  pThrd = item->pThrd;
  SSvrConn*   conn = NULL;
  queue       wq;
dengyihao's avatar
dengyihao 已提交
487

dengyihao's avatar
dengyihao 已提交
488
  // batch process to avoid to lock/unlock frequently
wafwerar's avatar
wafwerar 已提交
489
  taosThreadMutexLock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
490
  QUEUE_MOVE(&item->qmsg, &wq);
wafwerar's avatar
wafwerar 已提交
491
  taosThreadMutexUnlock(&item->mtx);
dengyihao's avatar
dengyihao 已提交
492 493 494 495

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

dengyihao's avatar
dengyihao 已提交
497
    SSvrMsg* msg = QUEUE_DATA(head, SSvrMsg, q);
dengyihao's avatar
dengyihao 已提交
498
    if (msg == NULL) {
dengyihao's avatar
dengyihao 已提交
499
      tError("unexcept occurred, continue");
dengyihao's avatar
dengyihao 已提交
500
      continue;
dengyihao's avatar
dengyihao 已提交
501
    }
dengyihao's avatar
dengyihao 已提交
502
    // release handle to rpc init
dengyihao's avatar
dengyihao 已提交
503 504
    if (msg->type == Quit) {
      (*transAsyncHandle[msg->type])(msg, pThrd);
dengyihao's avatar
dengyihao 已提交
505
      continue;
dengyihao's avatar
dengyihao 已提交
506 507 508
    } else {
      STransMsg transMsg = msg->msg;

S
Shengliang Guan 已提交
509 510
      SExHandle* exh1 = transMsg.info.handle;
      int64_t    refId = transMsg.info.refId;
dengyihao's avatar
dengyihao 已提交
511
      SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
512
      if (exh2 == NULL || exh1 != exh2) {
dengyihao's avatar
dengyihao 已提交
513
        tTrace("handle except msg %p, ignore it", exh1);
dengyihao's avatar
dengyihao 已提交
514
        transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
515 516 517 518
        destroySmsg(msg);
        continue;
      }
      msg->pConn = exh1->handle;
dengyihao's avatar
dengyihao 已提交
519
      transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
520
      (*transAsyncHandle[msg->type])(msg, pThrd);
dengyihao's avatar
dengyihao 已提交
521
    }
dengyihao's avatar
dengyihao 已提交
522 523
  }
}
dengyihao's avatar
dengyihao 已提交
524 525 526 527 528
static void uvWalkCb(uv_handle_t* handle, void* arg) {
  if (!uv_is_closing(handle)) {
    uv_close(handle, NULL);
  }
}
dengyihao's avatar
dengyihao 已提交
529 530 531 532
static void uvFreeCb(uv_handle_t* handle) {
  //
  taosMemoryFree(handle);
}
dengyihao's avatar
dengyihao 已提交
533

dengyihao's avatar
dengyihao 已提交
534 535
static void uvAcceptAsyncCb(uv_async_t* async) {
  SServerObj* srv = async->data;
dengyihao's avatar
dengyihao 已提交
536
  tDebug("close server port %d", srv->port);
dengyihao's avatar
dengyihao 已提交
537
  uv_walk(srv->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
538
}
dengyihao's avatar
dengyihao 已提交
539

dengyihao's avatar
dengyihao 已提交
540
static void uvShutDownCb(uv_shutdown_t* req, int status) {
dengyihao's avatar
dengyihao 已提交
541
  if (status != 0) {
S
Shengliang Guan 已提交
542
    tDebug("conn failed to shut down:%s", uv_err_name(status));
dengyihao's avatar
dengyihao 已提交
543
  }
dengyihao's avatar
dengyihao 已提交
544
  uv_close((uv_handle_t*)req->handle, uvDestroyConn);
wafwerar's avatar
wafwerar 已提交
545
  taosMemoryFree(req);
dengyihao's avatar
dengyihao 已提交
546 547
}

dengyihao's avatar
dengyihao 已提交
548 549 550
static void uvWorkDoTask(uv_work_t* req) {
  // doing time-consumeing task
  // only auth conn currently, add more func later
dengyihao's avatar
dengyihao 已提交
551
  tTrace("conn %p start to be processed in BG Thread", req->data);
dengyihao's avatar
dengyihao 已提交
552 553 554 555 556
  return;
}

static void uvWorkAfterTask(uv_work_t* req, int status) {
  if (status != 0) {
dengyihao's avatar
dengyihao 已提交
557
    tTrace("conn %p failed to processed ", req->data);
dengyihao's avatar
dengyihao 已提交
558 559 560 561
  }
  // Done time-consumeing task
  // add more func later
  // this func called in main loop
dengyihao's avatar
dengyihao 已提交
562
  tTrace("conn %p already processed ", req->data);
dengyihao's avatar
dengyihao 已提交
563 564 565
  taosMemoryFree(req);
}

dengyihao's avatar
dengyihao 已提交
566 567 568 569 570 571
void uvOnAcceptCb(uv_stream_t* stream, int status) {
  if (status == -1) {
    return;
  }
  SServerObj* pObj = container_of(stream, SServerObj, server);

wafwerar's avatar
wafwerar 已提交
572
  uv_tcp_t* cli = (uv_tcp_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
573 574 575
  uv_tcp_init(pObj->loop, cli);

  if (uv_accept(stream, (uv_stream_t*)cli) == 0) {
wafwerar's avatar
wafwerar 已提交
576
    if (pObj->numOfWorkerReady < pObj->numOfThreads) {
dengyihao's avatar
dengyihao 已提交
577 578
      tError("worker-threads are not ready for all, need %d instead of %d.", pObj->numOfThreads,
             pObj->numOfWorkerReady);
wafwerar's avatar
wafwerar 已提交
579 580 581
      uv_close((uv_handle_t*)cli, NULL);
      return;
    }
dengyihao's avatar
dengyihao 已提交
582

wafwerar's avatar
wafwerar 已提交
583
    uv_write_t* wr = (uv_write_t*)taosMemoryMalloc(sizeof(uv_write_t));
dengyihao's avatar
dengyihao 已提交
584
    wr->data = cli;
dengyihao's avatar
dengyihao 已提交
585 586 587
    uv_buf_t buf = uv_buf_init((char*)notify, strlen(notify));

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

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

dengyihao's avatar
dengyihao 已提交
591
    uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb);
dengyihao's avatar
dengyihao 已提交
592 593 594 595 596
  } else {
    uv_close((uv_handle_t*)cli, NULL);
  }
}
void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
dengyihao's avatar
dengyihao 已提交
597
  tTrace("connection coming");
dengyihao's avatar
dengyihao 已提交
598 599 600 601 602
  if (nread < 0) {
    if (nread != UV_EOF) {
      tError("read error %s", uv_err_name(nread));
    }
    // TODO(log other failure reason)
S
Shengliang Guan 已提交
603
    tWarn("failed to create connect:%p", q);
dengyihao's avatar
dengyihao 已提交
604 605 606
    taosMemoryFree(buf->base);
    uv_close((uv_handle_t*)q, NULL);
    // taosMemoryFree(q);
dengyihao's avatar
dengyihao 已提交
607 608 609 610 611
    return;
  }
  // free memory allocated by
  assert(nread == strlen(notify));
  assert(buf->base[0] == notify[0]);
wafwerar's avatar
wafwerar 已提交
612
  taosMemoryFree(buf->base);
dengyihao's avatar
dengyihao 已提交
613

dengyihao's avatar
dengyihao 已提交
614
  SWorkThrd* pThrd = q->data;
dengyihao's avatar
dengyihao 已提交
615 616 617 618 619 620 621 622 623 624

  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);
  assert(pending == UV_TCP);

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

dengyihao's avatar
dengyihao 已提交
627
  pConn->pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
628
  /* init conn timer*/
dengyihao's avatar
dengyihao 已提交
629 630
  // uv_timer_init(pThrd->loop, &pConn->pTimer);
  // pConn->pTimer.data = pConn;
dengyihao's avatar
dengyihao 已提交
631 632 633 634

  pConn->hostThrd = pThrd;

  // init client handle
wafwerar's avatar
wafwerar 已提交
635
  pConn->pTcp = (uv_tcp_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
dengyihao's avatar
dengyihao 已提交
636 637 638
  uv_tcp_init(pThrd->loop, pConn->pTcp);
  pConn->pTcp->data = pConn;

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

dengyihao's avatar
dengyihao 已提交
641 642 643
  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 已提交
644
    tTrace("conn %p created, fd:%d", pConn, fd);
dengyihao's avatar
dengyihao 已提交
645

dengyihao's avatar
dengyihao 已提交
646 647
    int addrlen = sizeof(pConn->addr);
    if (0 != uv_tcp_getpeername(pConn->pTcp, (struct sockaddr*)&pConn->addr, &addrlen)) {
dengyihao's avatar
dengyihao 已提交
648
      tError("conn %p failed to get peer info", pConn);
dengyihao's avatar
dengyihao 已提交
649
      transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
650
      return;
dengyihao's avatar
dengyihao 已提交
651
    }
dengyihao's avatar
dengyihao 已提交
652

dengyihao's avatar
dengyihao 已提交
653 654
    addrlen = sizeof(pConn->localAddr);
    if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&pConn->localAddr, &addrlen)) {
dengyihao's avatar
dengyihao 已提交
655
      tError("conn %p failed to get local info", pConn);
dengyihao's avatar
dengyihao 已提交
656
      transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
657 658 659
      return;
    }

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

dengyihao's avatar
dengyihao 已提交
662
  } else {
dengyihao's avatar
dengyihao 已提交
663
    tDebug("failed to create new connection");
dengyihao's avatar
dengyihao 已提交
664
    transUnrefSrvHandle(pConn);
dengyihao's avatar
dengyihao 已提交
665 666 667
  }
}

668
void* transAcceptThread(void* arg) {
dengyihao's avatar
dengyihao 已提交
669
  // opt
dengyihao's avatar
dengyihao 已提交
670
  setThreadName("trans-accept");
dengyihao's avatar
dengyihao 已提交
671 672
  SServerObj* srv = (SServerObj*)arg;
  uv_run(srv->loop, UV_RUN_DEFAULT);
673 674

  return NULL;
dengyihao's avatar
dengyihao 已提交
675
}
dengyihao's avatar
dengyihao 已提交
676
void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
wafwerar's avatar
wafwerar 已提交
677 678 679
  if (status != 0) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
680
  SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
wafwerar's avatar
wafwerar 已提交
681 682
  uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
}
dengyihao's avatar
dengyihao 已提交
683
static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
wafwerar's avatar
wafwerar 已提交
684
  pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
685 686 687
  if (0 != uv_loop_init(pThrd->loop)) {
    return false;
  }
dengyihao's avatar
dengyihao 已提交
688

dengyihao's avatar
dengyihao 已提交
689
  uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
wafwerar's avatar
wafwerar 已提交
690
  // int r = uv_pipe_open(pThrd->pipe, pThrd->fd);
dengyihao's avatar
dengyihao 已提交
691 692 693

  pThrd->pipe->data = pThrd;

dengyihao's avatar
dengyihao 已提交
694
  QUEUE_INIT(&pThrd->msg);
wafwerar's avatar
wafwerar 已提交
695
  taosThreadMutexInit(&pThrd->msgMtx, NULL);
dengyihao's avatar
dengyihao 已提交
696

dengyihao's avatar
fix bug  
dengyihao 已提交
697 698 699
  // conn set
  QUEUE_INIT(&pThrd->conn);

dengyihao's avatar
dengyihao 已提交
700
  pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 1, pThrd, uvWorkerAsyncCb);
wafwerar's avatar
wafwerar 已提交
701 702
  uv_pipe_connect(&pThrd->connect_req, pThrd->pipe, pipeName, uvOnPipeConnectionCb);
  // uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
dengyihao's avatar
dengyihao 已提交
703 704 705 706 707 708 709 710 711
  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 已提交
712
    tError("failed to init accept server:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
713 714 715
    return false;
  }

dengyihao's avatar
dengyihao 已提交
716
  // register an async here to quit server gracefully
wafwerar's avatar
wafwerar 已提交
717
  srv->pAcceptAsync = taosMemoryCalloc(1, sizeof(uv_async_t));
dengyihao's avatar
dengyihao 已提交
718 719
  uv_async_init(srv->loop, srv->pAcceptAsync, uvAcceptAsyncCb);
  srv->pAcceptAsync->data = srv;
dengyihao's avatar
dengyihao 已提交
720

dengyihao's avatar
dengyihao 已提交
721
  struct sockaddr_in bind_addr;
dengyihao's avatar
dengyihao 已提交
722 723
  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 已提交
724
    tError("failed to bind:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
725 726
    return false;
  }
dengyihao's avatar
dengyihao 已提交
727
  if ((err = uv_listen((uv_stream_t*)&srv->server, 512, uvOnAcceptCb)) != 0) {
S
Shengliang Guan 已提交
728
    tError("failed to listen:%s", uv_err_name(err));
dengyihao's avatar
dengyihao 已提交
729
    terrno = TSDB_CODE_RPC_PORT_EADDRINUSE;
dengyihao's avatar
dengyihao 已提交
730 731 732
    return false;
  }
  return true;
dengyihao's avatar
dengyihao 已提交
733
}
734
void* transWorkerThread(void* arg) {
dengyihao's avatar
dengyihao 已提交
735
  setThreadName("trans-worker");
dengyihao's avatar
dengyihao 已提交
736
  SWorkThrd* pThrd = (SWorkThrd*)arg;
dengyihao's avatar
dengyihao 已提交
737
  uv_run(pThrd->loop, UV_RUN_DEFAULT);
738 739

  return NULL;
dengyihao's avatar
dengyihao 已提交
740 741
}

dengyihao's avatar
dengyihao 已提交
742
static SSvrConn* createConn(void* hThrd) {
dengyihao's avatar
dengyihao 已提交
743
  SWorkThrd* pThrd = hThrd;
dengyihao's avatar
fix bug  
dengyihao 已提交
744

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

dengyihao's avatar
dengyihao 已提交
747
  transReqQueueInit(&pConn->wreqQueue);
dengyihao's avatar
fix bug  
dengyihao 已提交
748 749 750
  QUEUE_INIT(&pConn->queue);

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

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

dengyihao's avatar
dengyihao 已提交
754
  memset(&pConn->regArg, 0, sizeof(pConn->regArg));
dengyihao's avatar
dengyihao 已提交
755
  pConn->broken = false;
dengyihao's avatar
dengyihao 已提交
756
  pConn->status = ConnNormal;
dengyihao's avatar
dengyihao 已提交
757

dengyihao's avatar
dengyihao 已提交
758 759 760
  SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
  exh->handle = pConn;
  exh->pThrd = pThrd;
dengyihao's avatar
dengyihao 已提交
761 762
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
763

dengyihao's avatar
dengyihao 已提交
764
  STrans* pTransInst = pThrd->pTransInst;
dengyihao's avatar
dengyihao 已提交
765
  pConn->refId = exh->refId;
dengyihao's avatar
dengyihao 已提交
766
  transRefSrvHandle(pConn);
S
Shengliang Guan 已提交
767
  tTrace("%s handle %p, conn %p created, refId:%" PRId64, transLabel(pTransInst), exh, pConn, pConn->refId);
dengyihao's avatar
dengyihao 已提交
768 769
  return pConn;
}
dengyihao's avatar
dengyihao 已提交
770

dengyihao's avatar
dengyihao 已提交
771
static void destroyConn(SSvrConn* conn, bool clear) {
dengyihao's avatar
dengyihao 已提交
772 773 774
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
775

dengyihao's avatar
dengyihao 已提交
776
  transDestroyBuffer(&conn->readBuf);
777
  if (clear) {
dengyihao's avatar
dengyihao 已提交
778
    if (!uv_is_closing((uv_handle_t*)conn->pTcp)) {
dengyihao's avatar
dengyihao 已提交
779 780 781
      tTrace("conn %p to be destroyed", conn);
      uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn);
    }
782
  }
dengyihao's avatar
dengyihao 已提交
783
}
dengyihao's avatar
dengyihao 已提交
784
static void destroyConnRegArg(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
785 786 787 788 789
  if (conn->regArg.init == 1) {
    transFreeMsg(conn->regArg.msg.pCont);
    conn->regArg.init = 0;
  }
}
dengyihao's avatar
dengyihao 已提交
790
static int reallocConnRef(SSvrConn* conn) {
dengyihao's avatar
dengyihao 已提交
791 792
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
793 794 795 796
  // 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 已提交
797 798
  exh->refId = transAddExHandle(transGetRefMgt(), exh);
  transAcquireExHandle(transGetRefMgt(), exh->refId);
dengyihao's avatar
dengyihao 已提交
799 800 801 802
  conn->refId = exh->refId;

  return 0;
}
dengyihao's avatar
dengyihao 已提交
803
static void uvDestroyConn(uv_handle_t* handle) {
dengyihao's avatar
dengyihao 已提交
804
  SSvrConn* conn = handle->data;
dengyihao's avatar
dengyihao 已提交
805 806 807
  if (conn == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
808
  SWorkThrd* thrd = conn->hostThrd;
dengyihao's avatar
dengyihao 已提交
809

dengyihao's avatar
dengyihao 已提交
810 811
  transReleaseExHandle(transGetRefMgt(), conn->refId);
  transRemoveExHandle(transGetRefMgt(), conn->refId);
dengyihao's avatar
dengyihao 已提交
812

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

dengyihao's avatar
dengyihao 已提交
816 817 818 819
  for (int i = 0; i < transQueueSize(&conn->srvMsgs); i++) {
    SSvrMsg* msg = transQueueGet(&conn->srvMsgs, i);
    destroySmsg(msg);
  }
dengyihao's avatar
dengyihao 已提交
820

dengyihao's avatar
dengyihao 已提交
821
  transReqQueueClear(&conn->wreqQueue);
dengyihao's avatar
dengyihao 已提交
822
  transQueueDestroy(&conn->srvMsgs);
dengyihao's avatar
dengyihao 已提交
823

dengyihao's avatar
dengyihao 已提交
824
  QUEUE_REMOVE(&conn->queue);
wafwerar's avatar
wafwerar 已提交
825
  taosMemoryFree(conn->pTcp);
dengyihao's avatar
dengyihao 已提交
826
  destroyConnRegArg(conn);
dengyihao's avatar
dengyihao 已提交
827
  taosMemoryFree(conn);
dengyihao's avatar
dengyihao 已提交
828

dengyihao's avatar
dengyihao 已提交
829
  if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
dengyihao's avatar
dengyihao 已提交
830
    tTrace("work thread quit");
dengyihao's avatar
dengyihao 已提交
831
    uv_walk(thrd->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
832
  }
dengyihao's avatar
dengyihao 已提交
833
}
wafwerar's avatar
wafwerar 已提交
834 835 836 837
static void uvPipeListenCb(uv_stream_t* handle, int status) {
  ASSERT(status == 0);

  SServerObj* srv = container_of(handle, SServerObj, pipeListen);
dengyihao's avatar
dengyihao 已提交
838
  uv_pipe_t*  pipe = &(srv->pipe[srv->numOfWorkerReady][0]);
wafwerar's avatar
wafwerar 已提交
839 840 841 842 843 844 845 846 847 848 849 850 851 852
  ASSERT(0 == uv_pipe_init(srv->loop, pipe, 1));
  ASSERT(0 == uv_accept((uv_stream_t*)&srv->pipeListen, (uv_stream_t*)pipe));

  ASSERT(1 == uv_is_readable((uv_stream_t*)pipe));
  ASSERT(1 == uv_is_writable((uv_stream_t*)pipe));
  ASSERT(0 == uv_is_closing((uv_handle_t*)pipe));

  srv->numOfWorkerReady++;

  // ASSERT(0 == uv_listen((uv_stream_t*)&ctx.send.tcp, 512, uvOnAcceptCb));

  // r = uv_read_start((uv_stream_t*)&ctx.channel, alloc_cb, read_cb);
  // ASSERT(r == 0);
}
dengyihao's avatar
dengyihao 已提交
853

U
ubuntu 已提交
854
void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
wafwerar's avatar
wafwerar 已提交
855 856
  SServerObj* srv = taosMemoryCalloc(1, sizeof(SServerObj));
  srv->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
dengyihao's avatar
dengyihao 已提交
857 858
  srv->numOfThreads = numOfThreads;
  srv->workerIdx = 0;
wafwerar's avatar
wafwerar 已提交
859
  srv->numOfWorkerReady = 0;
dengyihao's avatar
dengyihao 已提交
860
  srv->pThreadObj = (SWorkThrd**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrd*));
wafwerar's avatar
wafwerar 已提交
861
  srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*));
dengyihao's avatar
dengyihao 已提交
862 863 864 865
  srv->ip = ip;
  srv->port = port;
  uv_loop_init(srv->loop);

wafwerar's avatar
wafwerar 已提交
866 867
  assert(0 == uv_pipe_init(srv->loop, &srv->pipeListen, 0));
#ifdef WINDOWS
868
  char pipeName[64];
S
Shengliang Guan 已提交
869
  snprintf(pipeName, sizeof(pipeName), "\\\\?\\pipe\\trans.rpc.%p-" PRIu64, taosSafeRand(), GetCurrentProcessId());
wafwerar's avatar
wafwerar 已提交
870
#else
871
  char pipeName[PATH_MAX] = {0};
S
Shengliang Guan 已提交
872
  snprintf(pipeName, sizeof(pipeName), "%s%spipe.trans.rpc.%08X-" PRIu64, tsTempDir, TD_DIRSEP, taosSafeRand(),
dengyihao's avatar
dengyihao 已提交
873
           taosGetSelfPthreadId());
wafwerar's avatar
wafwerar 已提交
874 875 876 877
#endif
  assert(0 == uv_pipe_bind(&srv->pipeListen, pipeName));
  assert(0 == uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb));

dengyihao's avatar
dengyihao 已提交
878
  for (int i = 0; i < srv->numOfThreads; i++) {
dengyihao's avatar
dengyihao 已提交
879
    SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));
dengyihao's avatar
dengyihao 已提交
880
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
881
    thrd->quit = false;
dengyihao's avatar
dengyihao 已提交
882
    srv->pThreadObj[i] = thrd;
dengyihao's avatar
dengyihao 已提交
883
    thrd->pTransInst = shandle;
dengyihao's avatar
dengyihao 已提交
884

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

dengyihao's avatar
dengyihao 已提交
888
    if (false == addHandleToWorkloop(thrd, pipeName)) {
dengyihao's avatar
dengyihao 已提交
889 890
      goto End;
    }
891
    int err = taosThreadCreate(&(thrd->thread), NULL, transWorkerThread, (void*)(thrd));
dengyihao's avatar
dengyihao 已提交
892
    if (err == 0) {
S
Shengliang Guan 已提交
893
      tDebug("success to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
894 895
    } else {
      // TODO: clear all other resource later
S
Shengliang Guan 已提交
896
      tError("failed to create worker-thread:%d", i);
dengyihao's avatar
dengyihao 已提交
897
      goto End;
dengyihao's avatar
dengyihao 已提交
898 899
    }
  }
900
  if (false == taosValidIpAndPort(srv->ip, srv->port)) {
dengyihao's avatar
dengyihao 已提交
901
    terrno = TAOS_SYSTEM_ERROR(errno);
S
Shengliang Guan 已提交
902
    tError("invalid ip/port, %d:%d, reason:%s", srv->ip, srv->port, terrstr());
903 904
    goto End;
  }
dengyihao's avatar
dengyihao 已提交
905 906 907
  if (false == addHandleToAcceptloop(srv)) {
    goto End;
  }
908
  int err = taosThreadCreate(&srv->thread, NULL, transAcceptThread, (void*)srv);
dengyihao's avatar
dengyihao 已提交
909 910 911
  if (err == 0) {
    tDebug("success to create accept-thread");
  } else {
dengyihao's avatar
dengyihao 已提交
912 913
    tError("failed  to create accept-thread");
    goto End;
dengyihao's avatar
dengyihao 已提交
914 915
    // clear all resource later
  }
dengyihao's avatar
dengyihao 已提交
916
  srv->inited = true;
dengyihao's avatar
dengyihao 已提交
917
  return srv;
dengyihao's avatar
dengyihao 已提交
918
End:
U
ubuntu 已提交
919
  transCloseServer(srv);
dengyihao's avatar
dengyihao 已提交
920 921
  return NULL;
}
dengyihao's avatar
dengyihao 已提交
922

dengyihao's avatar
dengyihao 已提交
923
void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
924
  thrd->quit = true;
dengyihao's avatar
dengyihao 已提交
925
  if (QUEUE_IS_EMPTY(&thrd->conn)) {
dengyihao's avatar
dengyihao 已提交
926
    uv_walk(thrd->loop, uvWalkCb, NULL);
dengyihao's avatar
dengyihao 已提交
927 928 929
  } else {
    destroyAllConn(thrd);
  }
wafwerar's avatar
wafwerar 已提交
930
  taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
931
}
dengyihao's avatar
dengyihao 已提交
932
void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
933
  SSvrConn* conn = msg->pConn;
dengyihao's avatar
dengyihao 已提交
934
  if (conn->status == ConnAcquire) {
dengyihao's avatar
dengyihao 已提交
935
    reallocConnRef(conn);
dengyihao's avatar
dengyihao 已提交
936
    if (!transQueuePush(&conn->srvMsgs, msg)) {
dengyihao's avatar
dengyihao 已提交
937
      return;
dengyihao's avatar
dengyihao 已提交
938 939 940
    }
    uvStartSendRespInternal(msg);
    return;
dengyihao's avatar
dengyihao 已提交
941
  } else if (conn->status == ConnRelease || conn->status == ConnNormal) {
dengyihao's avatar
dengyihao 已提交
942
    tDebug("%s conn %p already released, ignore release-msg", transLabel(thrd->pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
943
  }
dengyihao's avatar
dengyihao 已提交
944
  destroySmsg(msg);
dengyihao's avatar
dengyihao 已提交
945
}
dengyihao's avatar
dengyihao 已提交
946
void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
947
  // send msg to client
dengyihao's avatar
dengyihao 已提交
948
  tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn);
dengyihao's avatar
dengyihao 已提交
949 950
  uvStartSendResp(msg);
}
dengyihao's avatar
dengyihao 已提交
951
void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
dengyihao's avatar
dengyihao 已提交
952
  SSvrConn* conn = msg->pConn;
dengyihao's avatar
dengyihao 已提交
953
  tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn);
dengyihao's avatar
dengyihao 已提交
954
  if (conn->status == ConnAcquire) {
dengyihao's avatar
dengyihao 已提交
955
    if (!transQueuePush(&conn->srvMsgs, msg)) {
dengyihao's avatar
dengyihao 已提交
956 957
      return;
    }
dengyihao's avatar
dengyihao 已提交
958
    transQueuePop(&conn->srvMsgs);
dengyihao's avatar
dengyihao 已提交
959 960 961
    conn->regArg.notifyCount = 0;
    conn->regArg.init = 1;
    conn->regArg.msg = msg->msg;
dengyihao's avatar
dengyihao 已提交
962
    tDebug("conn %p register brokenlink callback succ", conn);
dengyihao's avatar
dengyihao 已提交
963 964 965 966 967 968

    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 已提交
969
    taosMemoryFree(msg);
dengyihao's avatar
dengyihao 已提交
970 971
  }
}
dengyihao's avatar
dengyihao 已提交
972
void destroyWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
973 974 975
  if (pThrd == NULL) {
    return;
  }
wafwerar's avatar
wafwerar 已提交
976
  taosThreadJoin(pThrd->thread, NULL);
dengyihao's avatar
dengyihao 已提交
977
  SRV_RELEASE_UV(pThrd->loop);
dengyihao's avatar
dengyihao 已提交
978
  TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg);
dengyihao's avatar
dengyihao 已提交
979
  transDestroyAsyncPool(pThrd->asyncPool);
dengyihao's avatar
dengyihao 已提交
980
  taosMemoryFree(pThrd->loop);
wafwerar's avatar
wafwerar 已提交
981
  taosMemoryFree(pThrd);
dengyihao's avatar
dengyihao 已提交
982
}
dengyihao's avatar
dengyihao 已提交
983
void sendQuitToWorkThrd(SWorkThrd* pThrd) {
dengyihao's avatar
dengyihao 已提交
984
  SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
dengyihao's avatar
dengyihao 已提交
985
  msg->type = Quit;
dengyihao's avatar
dengyihao 已提交
986
  tDebug("server send quit msg to work thread");
dengyihao's avatar
dengyihao 已提交
987
  transAsyncSend(pThrd->asyncPool, &msg->q);
dengyihao's avatar
dengyihao 已提交
988 989
}

U
ubuntu 已提交
990
void transCloseServer(void* arg) {
dengyihao's avatar
dengyihao 已提交
991 992
  // impl later
  SServerObj* srv = arg;
dengyihao's avatar
dengyihao 已提交
993 994

  tDebug("send quit msg to accept thread");
dengyihao's avatar
dengyihao 已提交
995 996 997 998
  if (srv->inited) {
    uv_async_send(srv->pAcceptAsync);
    taosThreadJoin(srv->thread, NULL);
  }
dengyihao's avatar
dengyihao 已提交
999
  SRV_RELEASE_UV(srv->loop);
dengyihao's avatar
dengyihao 已提交
1000

dengyihao's avatar
dengyihao 已提交
1001 1002 1003 1004 1005
  for (int i = 0; i < srv->numOfThreads; i++) {
    sendQuitToWorkThrd(srv->pThreadObj[i]);
    destroyWorkThrd(srv->pThreadObj[i]);
  }

wafwerar's avatar
wafwerar 已提交
1006 1007 1008
  taosMemoryFree(srv->pThreadObj);
  taosMemoryFree(srv->pAcceptAsync);
  taosMemoryFree(srv->loop);
dengyihao's avatar
dengyihao 已提交
1009 1010

  for (int i = 0; i < srv->numOfThreads; i++) {
wafwerar's avatar
wafwerar 已提交
1011
    taosMemoryFree(srv->pipe[i]);
dengyihao's avatar
dengyihao 已提交
1012
  }
wafwerar's avatar
wafwerar 已提交
1013
  taosMemoryFree(srv->pipe);
dengyihao's avatar
dengyihao 已提交
1014

wafwerar's avatar
wafwerar 已提交
1015
  taosMemoryFree(srv);
dengyihao's avatar
dengyihao 已提交
1016
}
dengyihao's avatar
dengyihao 已提交
1017

dengyihao's avatar
dengyihao 已提交
1018 1019 1020 1021
void transRefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1022
  int ref = T_REF_INC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1023
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1024 1025 1026 1027 1028 1029
}

void transUnrefSrvHandle(void* handle) {
  if (handle == NULL) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1030
  int ref = T_REF_DEC((SSvrConn*)handle);
dengyihao's avatar
dengyihao 已提交
1031
  tTrace("conn %p ref count:%d", handle, ref);
dengyihao's avatar
dengyihao 已提交
1032
  if (ref == 0) {
dengyihao's avatar
dengyihao 已提交
1033
    destroyConn((SSvrConn*)handle, true);
dengyihao's avatar
dengyihao 已提交
1034 1035
  }
}
dengyihao's avatar
dengyihao 已提交
1036 1037

void transReleaseSrvHandle(void* handle) {
dengyihao's avatar
dengyihao 已提交
1038 1039 1040
  SRpcHandleInfo* info = handle;
  SExHandle*      exh = info->handle;
  int64_t         refId = info->refId;
dengyihao's avatar
dengyihao 已提交
1041

dengyihao's avatar
dengyihao 已提交
1042
  ASYNC_CHECK_HANDLE(exh, refId);
dengyihao's avatar
dengyihao 已提交
1043

dengyihao's avatar
dengyihao 已提交
1044
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1045 1046
  ASYNC_ERR_JRET(pThrd);

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

dengyihao's avatar
dengyihao 已提交
1049 1050 1051
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Release;
dengyihao's avatar
dengyihao 已提交
1052

dengyihao's avatar
dengyihao 已提交
1053
  tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
dengyihao's avatar
dengyihao 已提交
1054
  transAsyncSend(pThrd->asyncPool, &m->q);
dengyihao's avatar
dengyihao 已提交
1055
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1056
  return;
dengyihao's avatar
dengyihao 已提交
1057
_return1:
dengyihao's avatar
dengyihao 已提交
1058
  tTrace("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1059
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1060 1061
  return;
_return2:
dengyihao's avatar
dengyihao 已提交
1062
  tTrace("handle %p failed to send to release handle", exh);
dengyihao's avatar
dengyihao 已提交
1063
  return;
dengyihao's avatar
dengyihao 已提交
1064
}
dengyihao's avatar
dengyihao 已提交
1065
void transSendResponse(const STransMsg* msg) {
S
Shengliang Guan 已提交
1066 1067
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1068
  ASYNC_CHECK_HANDLE(exh, refId);
dengyihao's avatar
dengyihao 已提交
1069
  assert(refId != 0);
dengyihao's avatar
dengyihao 已提交
1070

dengyihao's avatar
dengyihao 已提交
1071
  STransMsg tmsg = *msg;
S
Shengliang Guan 已提交
1072
  tmsg.info.refId = refId;
dengyihao's avatar
dengyihao 已提交
1073

dengyihao's avatar
dengyihao 已提交
1074
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1075
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1076

dengyihao's avatar
dengyihao 已提交
1077 1078 1079
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Normal;
dengyihao's avatar
dengyihao 已提交
1080 1081 1082

  STraceId* trace = (STraceId*)&msg->info.traceId;
  tGTrace("conn %p start to send resp (1/2)", exh->handle);
dengyihao's avatar
dengyihao 已提交
1083
  transAsyncSend(pThrd->asyncPool, &m->q);
dengyihao's avatar
dengyihao 已提交
1084
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1085
  return;
dengyihao's avatar
dengyihao 已提交
1086
_return1:
dengyihao's avatar
dengyihao 已提交
1087
  tTrace("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1088
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1089
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1090 1091
  return;
_return2:
dengyihao's avatar
dengyihao 已提交
1092
  tTrace("handle %p failed to send resp", exh);
dengyihao's avatar
dengyihao 已提交
1093 1094
  rpcFreeCont(msg->pCont);
  return;
dengyihao's avatar
dengyihao 已提交
1095
}
dengyihao's avatar
dengyihao 已提交
1096
void transRegisterMsg(const STransMsg* msg) {
S
Shengliang Guan 已提交
1097 1098
  SExHandle* exh = msg->info.handle;
  int64_t    refId = msg->info.refId;
dengyihao's avatar
dengyihao 已提交
1099 1100
  ASYNC_CHECK_HANDLE(exh, refId);

dengyihao's avatar
dengyihao 已提交
1101
  STransMsg tmsg = *msg;
S
Shengliang Guan 已提交
1102
  tmsg.info.refId = refId;
dengyihao's avatar
dengyihao 已提交
1103

dengyihao's avatar
dengyihao 已提交
1104
  SWorkThrd* pThrd = exh->pThrd;
dengyihao's avatar
dengyihao 已提交
1105
  ASYNC_ERR_JRET(pThrd);
dengyihao's avatar
dengyihao 已提交
1106

dengyihao's avatar
dengyihao 已提交
1107 1108 1109
  SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
  m->msg = tmsg;
  m->type = Register;
dengyihao's avatar
dengyihao 已提交
1110

dengyihao's avatar
dengyihao 已提交
1111 1112
  STrans* pTransInst = pThrd->pTransInst;
  tTrace("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
dengyihao's avatar
dengyihao 已提交
1113
  transAsyncSend(pThrd->asyncPool, &m->q);
dengyihao's avatar
dengyihao 已提交
1114
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1115
  return;
dengyihao's avatar
dengyihao 已提交
1116

dengyihao's avatar
dengyihao 已提交
1117
_return1:
dengyihao's avatar
dengyihao 已提交
1118
  tTrace("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1119
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1120
  transReleaseExHandle(transGetRefMgt(), refId);
dengyihao's avatar
dengyihao 已提交
1121 1122
  return;
_return2:
dengyihao's avatar
dengyihao 已提交
1123
  tTrace("handle %p failed to register brokenlink", exh);
dengyihao's avatar
dengyihao 已提交
1124
  rpcFreeCont(msg->pCont);
dengyihao's avatar
dengyihao 已提交
1125
}
dengyihao's avatar
dengyihao 已提交
1126

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

dengyihao's avatar
dengyihao 已提交
1129
#endif