rpcTcp.c 16.9 KB
Newer Older
H
hzcheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

S
slguan 已提交
16
#include "os.h"
H
hzcheng 已提交
17 18
#include "tsocket.h"
#include "tutil.h"
19
#include "taosdef.h"
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
20
#include "taoserror.h" 
S
slguan 已提交
21
#include "rpcLog.h"
22
#include "rpcHead.h"
J
Jeff Tao 已提交
23
#include "rpcTcp.h"
S
TD-1057  
Shengliang Guan 已提交
24 25 26
#ifdef WINDOWS
#include "wepoll.h"
#endif
H
hzcheng 已提交
27

28 29 30
#ifndef EPOLLWAKEUP
  #define EPOLLWAKEUP (1u << 29)
#endif
H
hzcheng 已提交
31

J
Jeff Tao 已提交
32 33
typedef struct SFdObj {
  void              *signature;
S
TD-1057  
Shengliang Guan 已提交
34
  SOCKET             fd;          // TCP socket FD
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
35 36
  int                closedByApp; // 1: already closed by App
  void              *thandle;     // handle from upper layer, like TAOS
J
Jeff Tao 已提交
37 38 39 40 41
  uint32_t           ip;
  uint16_t           port;
  struct SThreadObj *pThreadObj;
  struct SFdObj     *prev;
  struct SFdObj     *next;
H
hzcheng 已提交
42 43
} SFdObj;

J
Jeff Tao 已提交
44
typedef struct SThreadObj {
H
hzcheng 已提交
45 46
  pthread_t       thread;
  SFdObj *        pHead;
J
Jeff Tao 已提交
47
  pthread_mutex_t mutex;
J
jtao1735 已提交
48
  uint32_t        ip;
49
  bool            stop;
S
TD-1057  
Shengliang Guan 已提交
50
  SOCKET          pollFd;
H
hzcheng 已提交
51 52
  int             numOfFds;
  int             threadId;
53
  char            label[TSDB_LABEL_LEN];
54 55
  void           *shandle;  // handle passed by upper layer during server initialization
  void           *(*processData)(SRecvInfo *pPacket);
H
hzcheng 已提交
56 57 58
} SThreadObj;

typedef struct {
S
TD-1057  
Shengliang Guan 已提交
59
  SOCKET      fd;
J
jtao1735 已提交
60
  uint32_t    ip;
L
lihui 已提交
61
  uint16_t    port;
62
  char        label[TSDB_LABEL_LEN];
H
hzcheng 已提交
63 64 65 66 67 68
  int         numOfThreads;
  void *      shandle;
  SThreadObj *pThreadObj;
  pthread_t   thread;
} SServerObj;

J
Jeff Tao 已提交
69
static void   *taosProcessTcpData(void *param);
S
TD-1057  
Shengliang Guan 已提交
70
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd);
J
Jeff Tao 已提交
71 72
static void    taosFreeFdObj(SFdObj *pFdObj);
static void    taosReportBrokenLink(SFdObj *pFdObj);
73
static void   *taosAcceptTcpConnection(void *arg);
H
hzcheng 已提交
74

J
jtao1735 已提交
75
void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle) {
J
Jeff Tao 已提交
76 77
  SServerObj *pServerObj;
  SThreadObj *pThreadObj;
H
hzcheng 已提交
78

J
Jeff Tao 已提交
79
  pServerObj = (SServerObj *)calloc(sizeof(SServerObj), 1);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
80 81 82 83 84 85
  if (pServerObj == NULL) {
    tError("TCP:%s no enough memory", label);
    terrno = TAOS_SYSTEM_ERROR(errno); 
    return NULL;
  }

86
  pServerObj->fd = -1;
87
  taosResetPthread(&pServerObj->thread);
J
jtao1735 已提交
88
  pServerObj->ip = ip;
89
  pServerObj->port = port;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
90
  tstrncpy(pServerObj->label, label, sizeof(pServerObj->label));
91 92
  pServerObj->numOfThreads = numOfThreads;

J
Jeff Tao 已提交
93
  pServerObj->pThreadObj = (SThreadObj *)calloc(sizeof(SThreadObj), numOfThreads);
94 95
  if (pServerObj->pThreadObj == NULL) {
    tError("TCP:%s no enough memory", label);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
96
    terrno = TAOS_SYSTEM_ERROR(errno); 
J
Jeff Tao 已提交
97
    free(pServerObj);
98
    return NULL;
H
hzcheng 已提交
99
  }
J
Jeff Tao 已提交
100

J
Jeff Tao 已提交
101
  int code = 0;
102 103 104 105
  pthread_attr_t thattr;
  pthread_attr_init(&thattr);
  pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);

106
  // initialize parameters in case it may encounter error later 
107
  pThreadObj = pServerObj->pThreadObj;
J
Jeff Tao 已提交
108
  for (int i = 0; i < numOfThreads; ++i) {
109
    pThreadObj->pollFd = -1;
110
    taosResetPthread(&pThreadObj->thread);
111
    pThreadObj->processData = fp;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
112
    tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
113
    pThreadObj->shandle = shandle;
114 115
    pThreadObj++;
  }
H
hzcheng 已提交
116

117 118 119
  // initialize mutex, thread, fd which may fail
  pThreadObj = pServerObj->pThreadObj;
  for (int i = 0; i < numOfThreads; ++i) {
J
Jeff Tao 已提交
120
    code = pthread_mutex_init(&(pThreadObj->mutex), NULL);
J
Jeff Tao 已提交
121 122
    if (code < 0) {
      tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno));
123
      break;
124
    }
H
hzcheng 已提交
125

S
TD-1057  
Shengliang Guan 已提交
126
    pThreadObj->pollFd = (int64_t)epoll_create(10);  // size does not matter
127 128
    if (pThreadObj->pollFd < 0) {
      tError("%s failed to create TCP epoll", label);
J
Jeff Tao 已提交
129 130
      code = -1;
      break;
131
    }
H
hzcheng 已提交
132

J
Jeff Tao 已提交
133
    code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
J
Jeff Tao 已提交
134 135 136
    if (code != 0) {
      tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
      break;
137
    }
H
hzcheng 已提交
138

139 140
    pThreadObj->threadId = i;
    pThreadObj++;
H
hzcheng 已提交
141 142
  }

143 144 145
  pServerObj->fd = taosOpenTcpServerSocket(pServerObj->ip, pServerObj->port);
  if (pServerObj->fd < 0) code = -1; 

J
Jeff Tao 已提交
146
  if (code == 0) { 
147
    code = pthread_create(&pServerObj->thread, &thattr, taosAcceptTcpConnection, (void *)pServerObj);
J
Jeff Tao 已提交
148 149 150
    if (code != 0) {
      tError("%s failed to create TCP accept thread(%s)", label, strerror(errno));
    }
H
hzcheng 已提交
151 152
  }

J
Jeff Tao 已提交
153
  if (code != 0) {
154
    terrno = TAOS_SYSTEM_ERROR(errno); 
155
    taosCleanUpTcpServer(pServerObj);
J
Jeff Tao 已提交
156 157
    pServerObj = NULL;
  } else {
158
    tDebug("%s TCP server is initialized, ip:0x%x port:%hu numOfThreads:%d", label, ip, port, numOfThreads);
J
Jeff Tao 已提交
159
  }
H
hzcheng 已提交
160

161
  pthread_attr_destroy(&thattr);
162
  return (void *)pServerObj;
H
hzcheng 已提交
163 164
}

165 166
static void taosStopTcpThread(SThreadObj* pThreadObj) {
  pThreadObj->stop = true;
167 168
  eventfd_t fd = -1;

S
TD-1057  
Shengliang Guan 已提交
169
  if (taosCheckPthreadValid(pThreadObj->thread) && pThreadObj->pollFd >= 0) {
170 171 172 173 174 175 176
    // signal the thread to stop, try graceful method first,
    // and use pthread_cancel when failed
    struct epoll_event event = { .events = EPOLLIN };
    fd = eventfd(1, 0);
    if (fd == -1) {
      // failed to create eventfd, call pthread_cancel instead, which may result in data corruption:
      tError("%s, failed to create eventfd(%s)", pThreadObj->label, strerror(errno));
S
TD-1057  
Shengliang Guan 已提交
177
      pThreadObj->stop = true;
178 179 180 181 182 183
      pthread_cancel(pThreadObj->thread);
    } else if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) {
      // failed to call epoll_ctl, call pthread_cancel instead, which may result in data corruption:
      tError("%s, failed to call epoll_ctl(%s)", pThreadObj->label, strerror(errno));
      pthread_cancel(pThreadObj->thread);
    }
184 185
  }

S
TD-1057  
Shengliang Guan 已提交
186 187 188
  if (taosCheckPthreadValid(pThreadObj->thread)) pthread_join(pThreadObj->thread, NULL);
  if (pThreadObj->pollFd >=0) taosCloseSocket(pThreadObj->pollFd);
  if (fd != -1) taosCloseSocket(fd);
189 190 191 192 193 194 195 196

  while (pThreadObj->pHead) {
    SFdObj *pFdObj = pThreadObj->pHead;
    pThreadObj->pHead = pFdObj->next;
    taosFreeFdObj(pFdObj);
  }
}

197
void taosStopTcpServer(void *handle) {
198
  SServerObj *pServerObj = handle;
H
hzcheng 已提交
199 200

  if (pServerObj == NULL) return;
201
  if(pServerObj->fd >=0) shutdown(pServerObj->fd, SHUT_RD);
S
TD-1057  
Shengliang Guan 已提交
202
  if (taosCheckPthreadValid(pServerObj->thread)) pthread_join(pServerObj->thread, NULL);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
203

204
  tDebug("%s TCP server is stopped", pServerObj->label);
205 206 207 208 209 210
}

void taosCleanUpTcpServer(void *handle) {
  SServerObj *pServerObj = handle;
  SThreadObj *pThreadObj;
  if (pServerObj == NULL) return;
H
hzcheng 已提交
211

J
Jeff Tao 已提交
212
  for (int i = 0; i < pServerObj->numOfThreads; ++i) {
H
hzcheng 已提交
213
    pThreadObj = pServerObj->pThreadObj + i;
214
    taosStopTcpThread(pThreadObj);
J
Jeff Tao 已提交
215
    pthread_mutex_destroy(&(pThreadObj->mutex));
H
hzcheng 已提交
216 217
  }

218
  tDebug("%s TCP server is cleaned up", pServerObj->label);
H
hzcheng 已提交
219

S
Shengliang Guan 已提交
220 221
  taosTFree(pServerObj->pThreadObj);
  taosTFree(pServerObj);
H
hzcheng 已提交
222 223
}

224
static void *taosAcceptTcpConnection(void *arg) {
S
TD-1057  
Shengliang Guan 已提交
225
  SOCKET             connFd = -1;
J
Jeff Tao 已提交
226 227 228 229 230 231
  struct sockaddr_in caddr;
  int                threadId = 0;
  SThreadObj        *pThreadObj;
  SServerObj        *pServerObj;

  pServerObj = (SServerObj *)arg;
232
  tDebug("%s TCP server is ready, ip:0x%x:%hu", pServerObj->label, pServerObj->ip, pServerObj->port);
J
Jeff Tao 已提交
233 234 235

  while (1) {
    socklen_t addrlen = sizeof(caddr);
236 237 238
    connFd = accept(pServerObj->fd, (struct sockaddr *)&caddr, &addrlen);
    if (connFd == -1) {
      if (errno == EINVAL) {
239
        tDebug("%s TCP server stop accepting new connections, exiting", pServerObj->label);
240 241
        break;
      }
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
242

243
      tError("%s TCP accept failure(%s)", pServerObj->label, strerror(errno));
J
Jeff Tao 已提交
244 245 246 247 248 249 250 251 252 253 254
      continue;
    }

    taosKeepTcpAlive(connFd);

    // pick up the thread to handle this connection
    pThreadObj = pServerObj->pThreadObj + threadId;

    SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd);
    if (pFdObj) {
      pFdObj->ip = caddr.sin_addr.s_addr;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
255
      pFdObj->port = htons(caddr.sin_port);
256
      tDebug("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label, 
257
              taosInetNtoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds);
J
Jeff Tao 已提交
258
    } else {
S
TD-1057  
Shengliang Guan 已提交
259
      taosCloseSocket(connFd);
260
      tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
261
             taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
J
Jeff Tao 已提交
262 263 264 265 266 267
    }  

    // pick up next thread for next connection
    threadId++;
    threadId = threadId % pServerObj->numOfThreads;
  }
268

S
TD-1057  
Shengliang Guan 已提交
269
  taosCloseSocket(pServerObj->fd);
270
  return NULL;
J
Jeff Tao 已提交
271 272
}

J
jtao1735 已提交
273
void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle) {
J
Jeff Tao 已提交
274 275 276 277 278
  SThreadObj    *pThreadObj;
  pthread_attr_t thattr;

  pThreadObj = (SThreadObj *)malloc(sizeof(SThreadObj));
  memset(pThreadObj, 0, sizeof(SThreadObj));
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
279
  tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
J
jtao1735 已提交
280
  pThreadObj->ip = ip;
J
Jeff Tao 已提交
281 282 283 284
  pThreadObj->shandle = shandle;

  if (pthread_mutex_init(&(pThreadObj->mutex), NULL) < 0) {
    tError("%s failed to init TCP client mutex(%s)", label, strerror(errno));
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
285
    free(pThreadObj);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
286
    terrno = TAOS_SYSTEM_ERROR(errno); 
J
Jeff Tao 已提交
287 288 289
    return NULL;
  }

S
TD-1057  
Shengliang Guan 已提交
290
  pThreadObj->pollFd = (SOCKET)epoll_create(10);  // size does not matter
J
Jeff Tao 已提交
291 292
  if (pThreadObj->pollFd < 0) {
    tError("%s failed to create TCP client epoll", label);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
293
    free(pThreadObj);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
294
    terrno = TAOS_SYSTEM_ERROR(errno); 
J
Jeff Tao 已提交
295 296 297 298 299 300 301 302 303 304
    return NULL;
  }

  pThreadObj->processData = fp;

  pthread_attr_init(&thattr);
  pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
  int code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
  pthread_attr_destroy(&thattr);
  if (code != 0) {
S
TD-1057  
Shengliang Guan 已提交
305
    taosCloseSocket(pThreadObj->pollFd);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
306
    free(pThreadObj);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
307
    terrno = TAOS_SYSTEM_ERROR(errno); 
J
Jeff Tao 已提交
308 309 310 311
    tError("%s failed to create TCP read data thread(%s)", label, strerror(errno));
    return NULL;
  }

312
  tDebug("%s TCP client is initialized, ip:%u:%hu", label, ip, port);
J
Jeff Tao 已提交
313 314 315 316

  return pThreadObj;
}

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
317 318 319 320
void taosStopTcpClient(void *chandle) {
  SThreadObj *pThreadObj = chandle;
  if (pThreadObj == NULL) return;

321
  tDebug ("%s TCP client is stopped", pThreadObj->label);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
322 323
}

J
Jeff Tao 已提交
324 325 326 327
void taosCleanUpTcpClient(void *chandle) {
  SThreadObj *pThreadObj = chandle;
  if (pThreadObj == NULL) return;

328
  taosStopTcpThread(pThreadObj);
329
  tDebug ("%s TCP client is cleaned up", pThreadObj->label);
J
Jeff Tao 已提交
330

S
Shengliang Guan 已提交
331
  taosTFree(pThreadObj);
J
Jeff Tao 已提交
332 333
}

J
jtao1735 已提交
334
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) {
J
Jeff Tao 已提交
335 336
  SThreadObj *    pThreadObj = shandle;

S
TD-1057  
Shengliang Guan 已提交
337
  SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
338
  if (fd < 0) return NULL;
J
Jeff Tao 已提交
339

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
340 341 342 343 344 345 346 347
  struct sockaddr_in sin;
  uint16_t localPort = 0;
  unsigned int addrlen = sizeof(sin);
  if (getsockname(fd, (struct sockaddr *)&sin, &addrlen) == 0 &&
      sin.sin_family == AF_INET && addrlen == sizeof(sin)) {
    localPort = (uint16_t)ntohs(sin.sin_port);
  }

J
Jeff Tao 已提交
348 349 350 351 352
  SFdObj *pFdObj = taosMallocFdObj(pThreadObj, fd);
  
  if (pFdObj) {
    pFdObj->thandle = thandle;
    pFdObj->port = port;
J
jtao1735 已提交
353
    pFdObj->ip = ip;
354
    tDebug("%s %p TCP connection to 0x%x:%hu is created, localPort:%hu FD:%p numOfFds:%d", 
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
355
            pThreadObj->label, thandle, ip, port, localPort, pFdObj, pThreadObj->numOfFds);
J
Jeff Tao 已提交
356
  } else {
S
TD-1057  
Shengliang Guan 已提交
357
    taosCloseSocket(fd);
J
Jeff Tao 已提交
358 359 360 361 362 363 364
    tError("%s failed to malloc client FdObj(%s)", pThreadObj->label, strerror(errno));
  }

  return pFdObj;
}

void taosCloseTcpConnection(void *chandle) {
365
  SFdObj *pFdObj = chandle;
366 367
  if (pFdObj == NULL) return;

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
368
  SThreadObj *pThreadObj = pFdObj->pThreadObj;
369
  tDebug("%s %p TCP connection will be closed, FD:%p", pThreadObj->label, pFdObj->thandle, pFdObj); 
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
370 371

  // pFdObj->thandle = NULL;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
372 373
  pFdObj->closedByApp = 1;
  shutdown(pFdObj->fd, SHUT_WR);
374 375
}

J
Jeff Tao 已提交
376
int taosSendTcpData(uint32_t ip, uint16_t port, void *data, int len, void *chandle) {
377
  SFdObj *pFdObj = chandle;
378 379 380

  if (chandle == NULL) return -1;

H
Haojun Liao 已提交
381
  return taosWriteMsg(pFdObj->fd, data, len);
382 383
}

384 385 386 387 388
static void taosReportBrokenLink(SFdObj *pFdObj) {

  SThreadObj *pThreadObj = pFdObj->pThreadObj;

  // notify the upper layer, so it will clean the associated context
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
389 390 391
  if (pFdObj->closedByApp == 0) {
    shutdown(pFdObj->fd, SHUT_WR);

392 393 394 395 396 397
    SRecvInfo recvInfo;
    recvInfo.msg = NULL;
    recvInfo.msgLen = 0;
    recvInfo.ip = 0;
    recvInfo.port = 0;
    recvInfo.shandle = pThreadObj->shandle;
398
    recvInfo.thandle = pFdObj->thandle;
399 400 401
    recvInfo.chandle = NULL;
    recvInfo.connType = RPC_CONN_TCP;
    (*(pThreadObj->processData))(&recvInfo);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
402 403 404 405 406 407 408 409 410 411 412 413 414 415
  } 

  taosFreeFdObj(pFdObj);
}

static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
  SRpcHead    rpcHead;
  int32_t     msgLen, leftLen, retLen, headLen;
  char       *buffer, *msg;

  SThreadObj *pThreadObj = pFdObj->pThreadObj;

  headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead));
  if (headLen != sizeof(SRpcHead)) {
416
    tDebug("%s %p read error, headLen:%d", pThreadObj->label, pFdObj->thandle, headLen);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
417 418 419 420 421 422
    return -1; 
  }

  msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
  buffer = malloc(msgLen + tsRpcOverhead);
  if ( NULL == buffer) {
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
423
    tError("%s %p TCP malloc(size:%d) fail", pThreadObj->label, pFdObj->thandle, msgLen);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
424
    return -1;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
425
  } else {
426
    tDebug("TCP malloc mem: %p", buffer);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
427 428 429 430 431 432 433
  }

  msg = buffer + tsRpcOverhead;
  leftLen = msgLen - headLen;
  retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen);

  if (leftLen != retLen) {
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
434 435
    tError("%s %p read error, leftLen:%d retLen:%d FD:%p", 
            pThreadObj->label, pFdObj->thandle, leftLen, retLen, pFdObj);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
436 437
    free(buffer);
    return -1;
438
  }
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
439 440 441 442 443 444 445 446

  memcpy(msg, &rpcHead, sizeof(SRpcHead));
  
  pInfo->msg = msg;
  pInfo->msgLen = msgLen;
  pInfo->ip = pFdObj->ip;
  pInfo->port = pFdObj->port;
  pInfo->shandle = pThreadObj->shandle;
447
  pInfo->thandle = pFdObj->thandle;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
448 449 450 451 452 453 454 455 456
  pInfo->chandle = pFdObj;
  pInfo->connType = RPC_CONN_TCP;

  if (pFdObj->closedByApp) {
    free(buffer); 
    return -1;
  }

  return 0;
457 458
}

J
Jeff Tao 已提交
459 460 461 462 463
#define maxEvents 10

static void *taosProcessTcpData(void *param) {
  SThreadObj        *pThreadObj = param;
  SFdObj            *pFdObj;
H
hzcheng 已提交
464
  struct epoll_event events[maxEvents];
465
  SRecvInfo          recvInfo;
466
 
H
hzcheng 已提交
467
  while (1) {
S
TD-1057  
Shengliang Guan 已提交
468
    int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME);
469
    if (pThreadObj->stop) {
470
      tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label);
471 472
      break;
    }
H
hzcheng 已提交
473 474
    if (fdNum < 0) continue;

J
Jeff Tao 已提交
475
    for (int i = 0; i < fdNum; ++i) {
H
hzcheng 已提交
476 477 478
      pFdObj = events[i].data.ptr;

      if (events[i].events & EPOLLERR) {
479
        tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj);
480
        taosReportBrokenLink(pFdObj);
H
hzcheng 已提交
481 482 483
        continue;
      }

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
484
      if (events[i].events & EPOLLRDHUP) {
485
        tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
486
        taosReportBrokenLink(pFdObj);
H
hzcheng 已提交
487 488 489
        continue;
      }

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
490
      if (events[i].events & EPOLLHUP) {
491
        tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
492
        taosReportBrokenLink(pFdObj);
493 494
        continue;
      }
H
hzcheng 已提交
495

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
496 497
      if (taosReadTcpData(pFdObj, &recvInfo) < 0) {
        shutdown(pFdObj->fd, SHUT_WR); 
H
hzcheng 已提交
498 499 500
        continue;
      }

501
      pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo);
502
      if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj);
H
hzcheng 已提交
503 504
    }
  }
J
Jeff Tao 已提交
505 506

  return NULL;
H
hzcheng 已提交
507 508
}

S
TD-1057  
Shengliang Guan 已提交
509
static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) {
H
hzcheng 已提交
510 511
  struct epoll_event event;

J
Jeff Tao 已提交
512
  SFdObj *pFdObj = (SFdObj *)calloc(sizeof(SFdObj), 1);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
513 514 515
  if (pFdObj == NULL) {
    return NULL;
  }
H
hzcheng 已提交
516

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
517
  pFdObj->closedByApp = 0;
J
Jeff Tao 已提交
518 519 520
  pFdObj->fd = fd;
  pFdObj->pThreadObj = pThreadObj;
  pFdObj->signature = pFdObj;
H
hzcheng 已提交
521

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
522
  event.events = EPOLLIN | EPOLLRDHUP;
J
Jeff Tao 已提交
523 524
  event.data.ptr = pFdObj;
  if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) {
S
Shengliang Guan 已提交
525
    taosTFree(pFdObj);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
526
    terrno = TAOS_SYSTEM_ERROR(errno); 
J
Jeff Tao 已提交
527
    return NULL;
H
hzcheng 已提交
528 529
  }

J
Jeff Tao 已提交
530 531 532 533 534 535 536
  // notify the data process, add into the FdObj list
  pthread_mutex_lock(&(pThreadObj->mutex));
  pFdObj->next = pThreadObj->pHead;
  if (pThreadObj->pHead) (pThreadObj->pHead)->prev = pFdObj;
  pThreadObj->pHead = pFdObj;
  pThreadObj->numOfFds++;
  pthread_mutex_unlock(&(pThreadObj->mutex));
H
hzcheng 已提交
537

J
Jeff Tao 已提交
538
  return pFdObj;
H
hzcheng 已提交
539 540
}

J
Jeff Tao 已提交
541
static void taosFreeFdObj(SFdObj *pFdObj) {
542 543 544 545

  if (pFdObj == NULL) return;
  if (pFdObj->signature != pFdObj) return;

546
  SThreadObj *pThreadObj = pFdObj->pThreadObj;
J
Jeff Tao 已提交
547
  pthread_mutex_lock(&pThreadObj->mutex);
548

J
Jeff Tao 已提交
549 550 551 552 553 554
  if (pFdObj->signature == NULL) {
    pthread_mutex_unlock(&pThreadObj->mutex);
    return;
  }

  pFdObj->signature = NULL;
555
  epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_DEL, pFdObj->fd, NULL);
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
556
  taosCloseSocket(pFdObj->fd);
557 558 559

  pThreadObj->numOfFds--;
  if (pThreadObj->numOfFds < 0)
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
560
    tError("%s %p TCP thread:%d, number of FDs is negative!!!", 
561
            pThreadObj->label, pFdObj->thandle, pThreadObj->threadId);
562 563 564 565 566 567 568 569 570 571 572

  if (pFdObj->prev) {
    (pFdObj->prev)->next = pFdObj->next;
  } else {
    pThreadObj->pHead = pFdObj->next;
  }

  if (pFdObj->next) {
    (pFdObj->next)->prev = pFdObj->prev;
  }

J
Jeff Tao 已提交
573
  pthread_mutex_unlock(&pThreadObj->mutex);
574

575
  tDebug("%s %p TCP connection is closed, FD:%p numOfFds:%d", 
576 577
          pThreadObj->label, pFdObj->thandle, pFdObj, pThreadObj->numOfFds);

S
Shengliang Guan 已提交
578
  taosTFree(pFdObj);
579
}