osSocket.c 28.5 KB
Newer Older
S
Shengliang Guan 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
17
#define ALLOW_FORBID_FUNC
S
Shengliang Guan 已提交
18
#include "os.h"
S
Shengliang Guan 已提交
19

20
#if defined(WINDOWS)
wafwerar's avatar
wafwerar 已提交
21 22 23 24 25 26 27
#include <IPHlpApi.h>
#include <WS2tcpip.h>
#include <Winsock2.h>
#include <stdio.h>
#include <string.h>
#include <tchar.h>
#include <winbase.h>
S
Shengliang Guan 已提交
28
#else
wafwerar's avatar
wafwerar 已提交
29 30
#include <arpa/inet.h>
#include <fcntl.h>
dengyihao's avatar
dengyihao 已提交
31
#include <net/if.h>
wafwerar's avatar
wafwerar 已提交
32 33 34 35 36 37 38
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/tcp.h>
#include <netinet/udp.h>
#include <sys/socket.h>
#include <unistd.h>
39 40

#if defined(DARWIN)
dengyihao's avatar
dengyihao 已提交
41 42
#include <dispatch/dispatch.h>
#include "osEok.h"
43
#else
dengyihao's avatar
dengyihao 已提交
44
#include <sys/epoll.h>
45
#endif
S
Shengliang Guan 已提交
46 47
#endif

dengyihao's avatar
dengyihao 已提交
48 49 50 51
#ifndef INVALID_SOCKET
#define INVALID_SOCKET -1
#endif

wafwerar's avatar
wafwerar 已提交
52 53
typedef struct TdSocketServer {
#if SOCKET_WITH_LOCK
wafwerar's avatar
wafwerar 已提交
54
  TdThreadRwlock rwlock;
wafwerar's avatar
wafwerar 已提交
55 56 57
#endif
  int      refId;
  SocketFd fd;
dengyihao's avatar
dengyihao 已提交
58
} * TdSocketServerPtr, TdSocketServer;
S
Shengliang Guan 已提交
59

wafwerar's avatar
wafwerar 已提交
60 61
typedef struct TdEpoll {
#if SOCKET_WITH_LOCK
wafwerar's avatar
wafwerar 已提交
62
  TdThreadRwlock rwlock;
wafwerar's avatar
wafwerar 已提交
63
#endif
dengyihao's avatar
dengyihao 已提交
64 65 66
  int     refId;
  EpollFd fd;
} * TdEpollPtr, TdEpoll;
S
Shengliang Guan 已提交
67

wafwerar's avatar
wafwerar 已提交
68
int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags, const struct sockaddr *dest_addr,
dengyihao's avatar
dengyihao 已提交
69
                   int addrlen) {
wafwerar's avatar
wafwerar 已提交
70 71
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
72
  }
wafwerar's avatar
wafwerar 已提交
73
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
74 75 76 77 78 79 80 81
  return sendto(pSocket->fd, buf, len, flags, dest_addr, addrlen);
#else
  return sendto(pSocket->fd, buf, len, flags, dest_addr, addrlen);
#endif
}
int32_t taosWriteSocket(TdSocketPtr pSocket, void *buf, int len) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
82
  }
wafwerar's avatar
wafwerar 已提交
83
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
84 85 86 87 88 89 90 91 92
  return send(pSocket->fd, buf, len, 0);
  ;
#else
  return write(pSocket->fd, buf, len);
#endif
}
int32_t taosReadSocket(TdSocketPtr pSocket, void *buf, int len) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
93
  }
wafwerar's avatar
wafwerar 已提交
94
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
95 96 97 98 99 100
  return recv(pSocket->fd, buf, len, 0);
  ;
#else
  return read(pSocket->fd, buf, len);
#endif
}
S
Shengliang Guan 已提交
101

dengyihao's avatar
dengyihao 已提交
102 103
int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr,
                           int *addrLen) {
wafwerar's avatar
wafwerar 已提交
104 105
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
106
  }
wafwerar's avatar
wafwerar 已提交
107 108 109
  return recvfrom(pSocket->fd, buf, len, flags, destAddr, addrLen);
}
int32_t taosCloseSocketNoCheck1(SocketFd fd) {
wafwerar's avatar
wafwerar 已提交
110
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
111 112 113
  return closesocket(fd);
#else
  return close(fd);
S
Shengliang Guan 已提交
114
#endif
wafwerar's avatar
wafwerar 已提交
115 116 117 118 119 120 121 122
}
int32_t taosCloseSocket(TdSocketPtr *ppSocket) {
  int32_t code;
  if (ppSocket == NULL || *ppSocket == NULL || (*ppSocket)->fd < 0) {
    return -1;
  }
  code = taosCloseSocketNoCheck1((*ppSocket)->fd);
  (*ppSocket)->fd = -1;
wafwerar's avatar
wafwerar 已提交
123
  taosMemoryFree(*ppSocket);
wafwerar's avatar
wafwerar 已提交
124 125 126 127 128 129 130 131 132
  return code;
}
int32_t taosCloseSocketServer(TdSocketServerPtr *ppSocketServer) {
  int32_t code;
  if (ppSocketServer == NULL || *ppSocketServer == NULL || (*ppSocketServer)->fd < 0) {
    return -1;
  }
  code = taosCloseSocketNoCheck1((*ppSocketServer)->fd);
  (*ppSocketServer)->fd = -1;
wafwerar's avatar
wafwerar 已提交
133
  taosMemoryFree(*ppSocketServer);
wafwerar's avatar
wafwerar 已提交
134 135
  return code;
}
S
Shengliang Guan 已提交
136

wafwerar's avatar
wafwerar 已提交
137 138 139 140
int32_t taosShutDownSocketRD(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
S
Shengliang Guan 已提交
141
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
142
  return closesocket(pSocket->fd);
S
Shengliang Guan 已提交
143
#elif __APPLE__
wafwerar's avatar
wafwerar 已提交
144
  return close(pSocket->fd);
S
Shengliang Guan 已提交
145
#else
wafwerar's avatar
wafwerar 已提交
146
  return shutdown(pSocket->fd, SHUT_RD);
S
Shengliang Guan 已提交
147 148
#endif
}
wafwerar's avatar
wafwerar 已提交
149 150 151 152
int32_t taosShutDownSocketServerRD(TdSocketServerPtr pSocketServer) {
  if (pSocketServer == NULL || pSocketServer->fd < 0) {
    return -1;
  }
S
Shengliang Guan 已提交
153
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
154
  return closesocket(pSocketServer->fd);
S
Shengliang Guan 已提交
155
#elif __APPLE__
wafwerar's avatar
wafwerar 已提交
156
  return close(pSocketServer->fd);
S
Shengliang Guan 已提交
157
#else
wafwerar's avatar
wafwerar 已提交
158
  return shutdown(pSocketServer->fd, SHUT_RD);
S
Shengliang Guan 已提交
159 160
#endif
}
S
Shengliang Guan 已提交
161

wafwerar's avatar
wafwerar 已提交
162 163 164
int32_t taosShutDownSocketWR(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
165
  }
wafwerar's avatar
wafwerar 已提交
166 167 168 169 170 171
#ifdef WINDOWS
  return closesocket(pSocket->fd);
#elif __APPLE__
  return close(pSocket->fd);
#else
  return shutdown(pSocket->fd, SHUT_WR);
S
slguan 已提交
172
#endif
S
Shengliang Guan 已提交
173
}
wafwerar's avatar
wafwerar 已提交
174 175 176 177 178 179 180 181 182 183
int32_t taosShutDownSocketServerWR(TdSocketServerPtr pSocketServer) {
  if (pSocketServer == NULL || pSocketServer->fd < 0) {
    return -1;
  }
#ifdef WINDOWS
  return closesocket(pSocketServer->fd);
#elif __APPLE__
  return close(pSocketServer->fd);
#else
  return shutdown(pSocketServer->fd, SHUT_WR);
S
Shengliang Guan 已提交
184
#endif
wafwerar's avatar
wafwerar 已提交
185 186 187 188
}
int32_t taosShutDownSocketRDWR(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
189
  }
wafwerar's avatar
wafwerar 已提交
190 191 192 193 194 195 196 197 198 199 200
#ifdef WINDOWS
  return closesocket(pSocket->fd);
#elif __APPLE__
  return close(pSocket->fd);
#else
  return shutdown(pSocket->fd, SHUT_RDWR);
#endif
}
int32_t taosShutDownSocketServerRDWR(TdSocketServerPtr pSocketServer) {
  if (pSocketServer == NULL || pSocketServer->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
201
  }
wafwerar's avatar
wafwerar 已提交
202 203 204 205 206 207 208
#ifdef WINDOWS
  return closesocket(pSocketServer->fd);
#elif __APPLE__
  return close(pSocketServer->fd);
#else
  return shutdown(pSocketServer->fd, SHUT_RDWR);
#endif
209
}
wafwerar's avatar
wafwerar 已提交
210

wafwerar's avatar
wafwerar 已提交
211 212
void taosWinSocketInit() {
#ifdef WINDOWS
S
Shengliang Guan 已提交
213 214 215 216 217 218 219 220 221
  static char flag = 0;
  if (flag == 0) {
    WORD    wVersionRequested;
    WSADATA wsaData;
    wVersionRequested = MAKEWORD(1, 1);
    if (WSAStartup(wVersionRequested, &wsaData) == 0) {
      flag = 1;
    }
  }
wafwerar's avatar
wafwerar 已提交
222 223
#else
#endif
224
}
wafwerar's avatar
wafwerar 已提交
225 226 227 228
int32_t taosSetNonblocking(TdSocketPtr pSocket, int32_t on) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
229
#ifdef WINDOWS
S
Shengliang Guan 已提交
230 231 232
  u_long mode;
  if (on) {
    mode = 1;
wafwerar's avatar
wafwerar 已提交
233
    ioctlsocket(pSocket->fd, FIONBIO, &mode);
S
Shengliang Guan 已提交
234 235
  } else {
    mode = 0;
wafwerar's avatar
wafwerar 已提交
236 237 238 239 240 241 242
    ioctlsocket(pSocket->fd, FIONBIO, &mode);
  }
#else
  int32_t flags = 0;
  if ((flags = fcntl(pSocket->fd, F_GETFL, 0)) < 0) {
    // printf("fcntl(F_GETFL) error: %d (%s)\n", errno, strerror(errno));
    return 1;
S
Shengliang Guan 已提交
243
  }
wafwerar's avatar
wafwerar 已提交
244 245 246 247 248 249 250 251 252 253 254

  if (on)
    flags |= O_NONBLOCK;
  else
    flags &= ~O_NONBLOCK;

  if ((flags = fcntl(pSocket->fd, F_SETFL, flags)) < 0) {
    // printf("fcntl(F_SETFL) error: %d (%s)\n", errno, strerror(errno));
    return 1;
  }
#endif
S
Shengliang Guan 已提交
255 256
  return 0;
}
wafwerar's avatar
wafwerar 已提交
257 258 259 260
int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t optlen) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
261
#ifdef WINDOWS
S
Shengliang Guan 已提交
262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277
  if (level == SOL_SOCKET && optname == TCP_KEEPCNT) {
    return 0;
  }

  if (level == SOL_TCP && optname == TCP_KEEPIDLE) {
    return 0;
  }

  if (level == SOL_TCP && optname == TCP_KEEPINTVL) {
    return 0;
  }

  if (level == SOL_TCP && optname == TCP_KEEPCNT) {
    return 0;
  }

wafwerar's avatar
wafwerar 已提交
278 279
  return setsockopt(pSocket->fd, level, optname, optval, optlen);
#else
wafwerar's avatar
wafwerar 已提交
280
  return setsockopt(pSocket->fd, level, optname, optval, (int)optlen);
wafwerar's avatar
wafwerar 已提交
281 282 283 284 285 286
#endif
}
int32_t taosGetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t *optlen) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
287
#ifdef WINDOWS
288
  assert(0);
wafwerar's avatar
wafwerar 已提交
289 290
  return 0;
#else
wafwerar's avatar
wafwerar 已提交
291
  return getsockopt(pSocket->fd, level, optname, optval, (int *)optlen);
wafwerar's avatar
wafwerar 已提交
292
#endif
S
Shengliang Guan 已提交
293
}
S
config  
Shengliang Guan 已提交
294
uint32_t taosInetAddr(const char *ipAddr) {
wafwerar's avatar
wafwerar 已提交
295
#ifdef WINDOWS
S
Shengliang Guan 已提交
296 297 298 299 300 301 302
  uint32_t value;
  int32_t  ret = inet_pton(AF_INET, ipAddr, &value);
  if (ret <= 0) {
    return INADDR_NONE;
  } else {
    return value;
  }
wafwerar's avatar
wafwerar 已提交
303 304 305
#else
  return inet_addr(ipAddr);
#endif
S
Shengliang Guan 已提交
306
}
S
Shengliang Guan 已提交
307
const char *taosInetNtoa(struct in_addr ipInt) {
wafwerar's avatar
wafwerar 已提交
308
#ifdef WINDOWS
S
Shengliang Guan 已提交
309 310 311
  // not thread safe, only for debug usage while print log
  static char tmpDstStr[16];
  return inet_ntop(AF_INET, &ipInt, tmpDstStr, INET6_ADDRSTRLEN);
wafwerar's avatar
wafwerar 已提交
312 313
#else
  return inet_ntoa(ipInt);
314
#endif
wafwerar's avatar
wafwerar 已提交
315
}
316 317

#ifndef SIGPIPE
wafwerar's avatar
wafwerar 已提交
318
#define SIGPIPE EPIPE
319 320 321 322
#endif

#define TCP_CONN_TIMEOUT 3000  // conn timeout

wafwerar's avatar
wafwerar 已提交
323 324 325 326
int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
327
  int32_t nleft, nwritten;
dengyihao's avatar
dengyihao 已提交
328
  char *  ptr = (char *)buf;
329 330 331 332

  nleft = nbytes;

  while (nleft > 0) {
wafwerar's avatar
wafwerar 已提交
333
    nwritten = taosWriteSocket(pSocket, (char *)ptr, (size_t)nleft);
334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351
    if (nwritten <= 0) {
      if (errno == EINTR /* || errno == EAGAIN || errno == EWOULDBLOCK */)
        continue;
      else
        return -1;
    } else {
      nleft -= nwritten;
      ptr += nwritten;
    }

    if (errno == SIGPIPE || errno == EPIPE) {
      return -1;
    }
  }

  return (nbytes - nleft);
}

wafwerar's avatar
wafwerar 已提交
352 353 354 355
int32_t taosReadMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
356
  int32_t nleft, nread;
dengyihao's avatar
dengyihao 已提交
357
  char *  ptr = (char *)buf;
358 359 360 361

  nleft = nbytes;

  while (nleft > 0) {
wafwerar's avatar
wafwerar 已提交
362
    nread = taosReadSocket(pSocket, ptr, (size_t)nleft);
363 364 365
    if (nread == 0) {
      break;
    } else if (nread < 0) {
wafwerar's avatar
wafwerar 已提交
366
      if (errno == EINTR /* || errno == EAGAIN || errno == EWOULDBLOCK*/) {
367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383
        continue;
      } else {
        return -1;
      }
    } else {
      nleft -= nread;
      ptr += nread;
    }

    if (errno == SIGPIPE || errno == EPIPE) {
      return -1;
    }
  }

  return (nbytes - nleft);
}

wafwerar's avatar
wafwerar 已提交
384 385 386 387 388
int32_t taosNonblockwrite(TdSocketPtr pSocket, char *ptr, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
  taosSetNonblocking(pSocket, 1);
389

wafwerar's avatar
wafwerar 已提交
390 391
  int32_t        nleft, nwritten, nready;
  fd_set         fset;
392 393 394 395 396 397 398
  struct timeval tv;

  nleft = nbytes;
  while (nleft > 0) {
    tv.tv_sec = 30;
    tv.tv_usec = 0;
    FD_ZERO(&fset);
wafwerar's avatar
wafwerar 已提交
399 400
    FD_SET(pSocket->fd, &fset);
    if ((nready = select((SocketFd)(pSocket->fd + 1), NULL, &fset, NULL, &tv)) == 0) {
401
      errno = ETIMEDOUT;
wafwerar's avatar
wafwerar 已提交
402
      // printf("fd %d timeout, no enough space to write", fd);
403 404 405 406 407
      break;

    } else if (nready < 0) {
      if (errno == EINTR) continue;

wafwerar's avatar
wafwerar 已提交
408
      // printf("select error, %d (%s)", errno, strerror(errno));
409 410 411
      return -1;
    }

wafwerar's avatar
wafwerar 已提交
412
    nwritten = (int32_t)send(pSocket->fd, ptr, (size_t)nleft, MSG_NOSIGNAL);
413 414 415
    if (nwritten <= 0) {
      if (errno == EAGAIN || errno == EINTR) continue;

wafwerar's avatar
wafwerar 已提交
416
      // printf("write error, %d (%s)", errno, strerror(errno));
417 418 419 420 421 422 423
      return -1;
    }

    nleft -= nwritten;
    ptr += nwritten;
  }

wafwerar's avatar
wafwerar 已提交
424
  taosSetNonblocking(pSocket, 0);
425 426 427 428

  return (nbytes - nleft);
}

wafwerar's avatar
wafwerar 已提交
429
TdSocketPtr taosOpenUdpSocket(uint32_t ip, uint16_t port) {
430
  struct sockaddr_in localAddr;
wafwerar's avatar
wafwerar 已提交
431 432
  SocketFd           fd;
  int32_t            bufSize = 1024000;
433

wafwerar's avatar
wafwerar 已提交
434
  // printf("open udp socket:0x%x:%hu", ip, port);
435 436 437 438 439 440

  memset((char *)&localAddr, 0, sizeof(localAddr));
  localAddr.sin_family = AF_INET;
  localAddr.sin_addr.s_addr = ip;
  localAddr.sin_port = (uint16_t)htons(port);

wafwerar's avatar
wafwerar 已提交
441 442 443 444
  if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) <= 2) {
    // printf("failed to open udp socket: %d (%s)", errno, strerror(errno));
    taosCloseSocketNoCheck1(fd);
    return NULL;
445 446
  }

wafwerar's avatar
wafwerar 已提交
447
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
448 449 450
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
451
  }
wafwerar's avatar
wafwerar 已提交
452 453
  pSocket->fd = fd;
  pSocket->refId = 0;
454

wafwerar's avatar
wafwerar 已提交
455 456 457 458 459 460 461 462 463 464
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
    // printf("failed to set the send buffer size for UDP socket\n");
    taosCloseSocket(&pSocket);
    return NULL;
  }

  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
    // printf("failed to set the receive buffer size for UDP socket\n");
    taosCloseSocket(&pSocket);
    return NULL;
465 466 467
  }

  /* bind socket to local address */
wafwerar's avatar
wafwerar 已提交
468 469 470 471
  if (bind(pSocket->fd, (struct sockaddr *)&localAddr, sizeof(localAddr)) < 0) {
    // printf("failed to bind udp socket: %d (%s), 0x%x:%hu", errno, strerror(errno), ip, port);
    taosCloseSocket(&pSocket);
    return NULL;
472 473
  }

wafwerar's avatar
wafwerar 已提交
474
  return pSocket;
475 476
}

wafwerar's avatar
wafwerar 已提交
477 478 479
TdSocketPtr taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientIp) {
  SocketFd           fd = -1;
  int32_t            ret;
480
  struct sockaddr_in serverAddr, clientAddr;
wafwerar's avatar
wafwerar 已提交
481
  int32_t            bufSize = 1024 * 1024;
482

wafwerar's avatar
wafwerar 已提交
483
  fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
484

wafwerar's avatar
wafwerar 已提交
485 486 487 488
  if (fd <= 2) {
    // printf("failed to open the socket: %d (%s)", errno, strerror(errno));
    if (fd >= 0) taosCloseSocketNoCheck1(fd);
    return NULL;
489 490
  }

wafwerar's avatar
wafwerar 已提交
491
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
492 493 494 495 496 497 498
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }
  pSocket->fd = fd;
  pSocket->refId = 0;

499 500
  /* set REUSEADDR option, so the portnumber can be re-used */
  int32_t reuse = 1;
wafwerar's avatar
wafwerar 已提交
501 502 503 504
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
    // printf("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
505 506
  }

wafwerar's avatar
wafwerar 已提交
507 508 509 510
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
    // printf("failed to set the send buffer size for TCP socket\n");
    taosCloseSocket(&pSocket);
    return NULL;
511 512
  }

wafwerar's avatar
wafwerar 已提交
513 514 515 516
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
    // printf("failed to set the receive buffer size for TCP socket\n");
    taosCloseSocket(&pSocket);
    return NULL;
517 518 519 520 521 522 523 524 525
  }

  if (clientIp != 0) {
    memset((char *)&clientAddr, 0, sizeof(clientAddr));
    clientAddr.sin_family = AF_INET;
    clientAddr.sin_addr.s_addr = clientIp;
    clientAddr.sin_port = 0;

    /* bind socket to client address */
wafwerar's avatar
wafwerar 已提交
526 527 528 529 530
    if (bind(pSocket->fd, (struct sockaddr *)&clientAddr, sizeof(clientAddr)) < 0) {
      // printf("bind tcp client socket failed, client(0x%x:0), dest(0x%x:%d), reason:(%s)", clientIp, destIp, destPort,
      //        strerror(errno));
      taosCloseSocket(&pSocket);
      return NULL;
531 532 533 534 535 536 537 538
    }
  }

  memset((char *)&serverAddr, 0, sizeof(serverAddr));
  serverAddr.sin_family = AF_INET;
  serverAddr.sin_addr.s_addr = destIp;
  serverAddr.sin_port = (uint16_t)htons((uint16_t)destPort);

wafwerar's avatar
wafwerar 已提交
539 540 541
#ifdef _TD_LINUX
  taosSetNonblocking(pSocket, 1);
  ret = connect(pSocket->fd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
542 543
  if (ret == -1) {
    if (errno == EHOSTUNREACH) {
wafwerar's avatar
wafwerar 已提交
544 545 546
      // printf("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
      taosCloseSocket(&pSocket);
      return -1;
547
    } else if (errno == EINPROGRESS || errno == EAGAIN || errno == EWOULDBLOCK) {
wafwerar's avatar
wafwerar 已提交
548
      struct pollfd wfd[1];
549

wafwerar's avatar
wafwerar 已提交
550
      wfd[0].fd = pSocket->fd;
551
      wfd[0].events = POLLOUT;
wafwerar's avatar
wafwerar 已提交
552

553 554
      int res = poll(wfd, 1, TCP_CONN_TIMEOUT);
      if (res == -1 || res == 0) {
wafwerar's avatar
wafwerar 已提交
555 556
        // printf("failed to connect socket, ip:0x%x, port:%hu(poll error/conn timeout)", destIp, destPort);
        taosCloseSocket(&pSocket);  //
557 558
        return -1;
      }
wafwerar's avatar
wafwerar 已提交
559 560 561 562
      int optVal = -1, optLen = sizeof(int);
      if ((0 != taosGetSockOpt(pSocket, SOL_SOCKET, SO_ERROR, &optVal, &optLen)) || (optVal != 0)) {
        // printf("failed to connect socket, ip:0x%x, port:%hu(connect host error)", destIp, destPort);
        taosCloseSocket(&pSocket);  //
563 564 565
        return -1;
      }
      ret = 0;
wafwerar's avatar
wafwerar 已提交
566 567 568 569 570
    } else {  // Other error
      // printf("failed to connect socket, ip:0x%x, port:%hu(target host cannot be reached)", destIp, destPort);
      taosCloseSocket(&pSocket);  //
      return -1;
    }
571
  }
wafwerar's avatar
wafwerar 已提交
572
  taosSetNonblocking(pSocket, 0);
573 574

#else
wafwerar's avatar
wafwerar 已提交
575
  ret = connect(pSocket->fd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
576 577 578
#endif

  if (ret != 0) {
wafwerar's avatar
wafwerar 已提交
579 580 581
    // printf("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
582
  } else {
wafwerar's avatar
wafwerar 已提交
583
    taosKeepTcpAlive(pSocket);
584 585
  }

wafwerar's avatar
wafwerar 已提交
586
  return pSocket;
587 588
}

wafwerar's avatar
wafwerar 已提交
589 590 591 592
int32_t taosKeepTcpAlive(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
593
  int32_t alive = 1;
wafwerar's avatar
wafwerar 已提交
594 595 596
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_KEEPALIVE, (void *)&alive, sizeof(alive)) < 0) {
    // printf("fd:%d setsockopt SO_KEEPALIVE failed: %d (%s)", sockFd, errno, strerror(errno));
    taosCloseSocket(&pSocket);
597 598 599 600 601 602
    return -1;
  }

#ifndef __APPLE__
  // all fails on macosx
  int32_t probes = 3;
wafwerar's avatar
wafwerar 已提交
603 604 605
  if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPCNT, (void *)&probes, sizeof(probes)) < 0) {
    // printf("fd:%d setsockopt SO_KEEPCNT failed: %d (%s)", sockFd, errno, strerror(errno));
    taosCloseSocket(&pSocket);
606 607 608 609
    return -1;
  }

  int32_t alivetime = 10;
wafwerar's avatar
wafwerar 已提交
610 611 612
  if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPIDLE, (void *)&alivetime, sizeof(alivetime)) < 0) {
    // printf("fd:%d setsockopt SO_KEEPIDLE failed: %d (%s)", sockFd, errno, strerror(errno));
    taosCloseSocket(&pSocket);
613 614 615 616
    return -1;
  }

  int32_t interval = 3;
wafwerar's avatar
wafwerar 已提交
617 618 619
  if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPINTVL, (void *)&interval, sizeof(interval)) < 0) {
    // printf("fd:%d setsockopt SO_KEEPINTVL failed: %d (%s)", sockFd, errno, strerror(errno));
    taosCloseSocket(&pSocket);
620 621
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
622
#endif  // __APPLE__
623 624

  int32_t nodelay = 1;
wafwerar's avatar
wafwerar 已提交
625 626 627
  if (taosSetSockOpt(pSocket, IPPROTO_TCP, TCP_NODELAY, (void *)&nodelay, sizeof(nodelay)) < 0) {
    // printf("fd:%d setsockopt TCP_NODELAY failed %d (%s)", sockFd, errno, strerror(errno));
    taosCloseSocket(&pSocket);
628 629 630 631 632 633
    return -1;
  }

  struct linger linger = {0};
  linger.l_onoff = 1;
  linger.l_linger = 3;
wafwerar's avatar
wafwerar 已提交
634 635 636
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_LINGER, (void *)&linger, sizeof(linger)) < 0) {
    // printf("setsockopt SO_LINGER failed: %d (%s)", errno, strerror(errno));
    taosCloseSocket(&pSocket);
637 638 639 640 641 642
    return -1;
  }

  return 0;
}

dengyihao's avatar
dengyihao 已提交
643 644 645
int taosGetLocalIp(const char *eth, char *ip) {
#if defined(WINDOWS)
  // DO NOTHAING
646
  assert(0);
dengyihao's avatar
dengyihao 已提交
647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672
  return 0;
#else
  int                fd;
  struct ifreq       ifr;
  struct sockaddr_in sin;

  fd = socket(AF_INET, SOCK_DGRAM, 0);
  if (-1 == fd) {
    return -1;
  }
  strncpy(ifr.ifr_name, eth, IFNAMSIZ);
  ifr.ifr_name[IFNAMSIZ - 1] = 0;

  if (ioctl(fd, SIOCGIFADDR, &ifr) < 0) {
    taosCloseSocketNoCheck1(fd);
    return -1;
  }
  memcpy(&sin, &ifr.ifr_addr, sizeof(sin));
  snprintf(ip, 64, "%s", inet_ntoa(sin.sin_addr));
  taosCloseSocketNoCheck1(fd);
#endif
  return 0;
}
int taosValidIp(uint32_t ip) {
#if defined(WINDOWS)
  // DO NOTHAING
673
  assert(0);
dengyihao's avatar
dengyihao 已提交
674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697
  return 0;
#else
  int ret = -1;
  int fd;

  struct ifconf ifconf;

  char buf[512] = {0};
  ifconf.ifc_len = 512;
  ifconf.ifc_buf = buf;

  if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
    return -1;
  }

  ioctl(fd, SIOCGIFCONF, &ifconf);
  struct ifreq *ifreq = (struct ifreq *)ifconf.ifc_buf;
  for (int i = (ifconf.ifc_len / sizeof(struct ifreq)); i > 0; i--) {
    char ip_str[64] = {0};
    if (ifreq->ifr_flags == AF_INET) {
      ret = taosGetLocalIp(ifreq->ifr_name, ip_str);
      if (ret != 0) {
        break;
      }
dengyihao's avatar
dengyihao 已提交
698 699 700 701 702
      ret = -1;
      if (ip == (uint32_t)taosInetAddr(ip_str)) {
        ret = 0;
        break;
      }
dengyihao's avatar
dengyihao 已提交
703 704 705 706 707 708 709 710 711
      ifreq++;
    }
  }
  taosCloseSocketNoCheck1(fd);
  return ret;
#endif
  return 0;
}

712 713 714 715 716 717 718 719 720
bool taosValidIpAndPort(uint32_t ip, uint16_t port) {
  struct sockaddr_in serverAdd;
  SocketFd           fd;
  int32_t            reuse;

  // printf("open tcp server socket:0x%x:%hu", ip, port);

  bzero((char *)&serverAdd, sizeof(serverAdd));
  serverAdd.sin_family = AF_INET;
wafwerar's avatar
wafwerar 已提交
721 722 723
#ifdef WINDOWS
  serverAdd.sin_addr.s_addr = INADDR_ANY;
#else
724
  serverAdd.sin_addr.s_addr = ip;
wafwerar's avatar
wafwerar 已提交
725
#endif
726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746
  serverAdd.sin_port = (uint16_t)htons(port);

  if ((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) <= 2) {
    // printf("failed to open TCP socket: %d (%s)", errno, strerror(errno));
    taosCloseSocketNoCheck1(fd);
    return false;
  }

  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return false;
  }
  pSocket->refId = 0;
  pSocket->fd = fd;

  /* set REUSEADDR option, so the portnumber can be re-used */
  reuse = 1;
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
    // printf("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
    taosCloseSocket(&pSocket);
dengyihao's avatar
dengyihao 已提交
747
    return false;
748 749 750 751 752 753 754 755
  }
  /* bind socket to server address */
  if (bind(pSocket->fd, (struct sockaddr *)&serverAdd, sizeof(serverAdd)) < 0) {
    // printf("bind tcp server socket failed, 0x%x:%hu(%s)", ip, port, strerror(errno));
    taosCloseSocket(&pSocket);
    return false;
  }
  taosCloseSocket(&pSocket);
dengyihao's avatar
dengyihao 已提交
756 757
  return true;
  // return 0 == taosValidIp(ip) ? true : false;
758
}
wafwerar's avatar
wafwerar 已提交
759
TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
760
  struct sockaddr_in serverAdd;
wafwerar's avatar
wafwerar 已提交
761
  SocketFd           fd;
762 763
  int32_t            reuse;

wafwerar's avatar
wafwerar 已提交
764
  // printf("open tcp server socket:0x%x:%hu", ip, port);
765 766 767 768 769 770

  bzero((char *)&serverAdd, sizeof(serverAdd));
  serverAdd.sin_family = AF_INET;
  serverAdd.sin_addr.s_addr = ip;
  serverAdd.sin_port = (uint16_t)htons(port);

wafwerar's avatar
wafwerar 已提交
771 772 773 774 775 776
  if ((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) <= 2) {
    // printf("failed to open TCP socket: %d (%s)", errno, strerror(errno));
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
777
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
778 779 780
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
781
  }
wafwerar's avatar
wafwerar 已提交
782 783
  pSocket->refId = 0;
  pSocket->fd = fd;
784 785 786

  /* set REUSEADDR option, so the portnumber can be re-used */
  reuse = 1;
wafwerar's avatar
wafwerar 已提交
787 788 789 790
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
    // printf("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
791 792 793
  }

  /* bind socket to server address */
wafwerar's avatar
wafwerar 已提交
794 795 796 797
  if (bind(pSocket->fd, (struct sockaddr *)&serverAdd, sizeof(serverAdd)) < 0) {
    // printf("bind tcp server socket failed, 0x%x:%hu(%s)", ip, port, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
798 799
  }

wafwerar's avatar
wafwerar 已提交
800 801 802 803
  if (taosKeepTcpAlive(pSocket) < 0) {
    // printf("failed to set tcp server keep-alive option, 0x%x:%hu(%s)", ip, port, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
804 805
  }

wafwerar's avatar
wafwerar 已提交
806 807 808 809
  if (listen(pSocket->fd, 1024) < 0) {
    // printf("listen tcp server socket failed, 0x%x:%hu(%s)", ip, port, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
810 811
  }

wafwerar's avatar
wafwerar 已提交
812
  return (TdSocketServerPtr)pSocket;
813 814
}

dengyihao's avatar
dengyihao 已提交
815
TdSocketPtr taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr, int *addrLen) {
wafwerar's avatar
wafwerar 已提交
816 817 818 819 820 821 822 823 824
  if (pServerSocket == NULL || pServerSocket->fd < 0) {
    return NULL;
  }
  SocketFd fd = accept(pServerSocket->fd, destAddr, addrLen);
  if (fd == -1) {
    // tError("TCP accept failure(%s)", strerror(errno));
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
825
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
826 827 828 829 830 831 832 833
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }
  pSocket->fd = fd;
  pSocket->refId = 0;
  return pSocket;
}
834 835 836
#define COPY_SIZE 32768
// sendfile shall be used

wafwerar's avatar
wafwerar 已提交
837 838 839 840
int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len) {
  if (pSrcSocket == NULL || pSrcSocket->fd < 0 || pDestSocket == NULL || pDestSocket->fd < 0) {
    return -1;
  }
841 842 843 844 845 846 847 848 849 850 851 852
  int64_t leftLen;
  int64_t readLen, writeLen;
  char    temp[COPY_SIZE];

  leftLen = len;

  while (leftLen > 0) {
    if (leftLen < COPY_SIZE)
      readLen = leftLen;
    else
      readLen = COPY_SIZE;  // 4K

wafwerar's avatar
wafwerar 已提交
853
    int64_t retLen = taosReadMsg(pSrcSocket, temp, (int32_t)readLen);
854
    if (readLen != retLen) {
wafwerar's avatar
wafwerar 已提交
855 856
      // printf("read error, readLen:%" PRId64 " retLen:%" PRId64 " len:%" PRId64 " leftLen:%" PRId64 ", reason:%s",
      //        readLen, retLen, len, leftLen, strerror(errno));
857 858 859
      return -1;
    }

wafwerar's avatar
wafwerar 已提交
860
    writeLen = taosWriteMsg(pDestSocket, temp, (int32_t)readLen);
861 862

    if (readLen != writeLen) {
wafwerar's avatar
wafwerar 已提交
863 864
      // printf("copy error, readLen:%" PRId64 " writeLen:%" PRId64 " len:%" PRId64 " leftLen:%" PRId64 ", reason:%s",
      //        readLen, writeLen, len, leftLen, strerror(errno));
865 866 867 868 869 870 871 872
      return -1;
    }

    leftLen -= readLen;
  }

  return len;
}
873 874

void taosBlockSIGPIPE() {
wafwerar's avatar
wafwerar 已提交
875
#ifdef WINDOWS
876
  // assert(0);
877
#else
878 879 880
  sigset_t signal_mask;
  sigemptyset(&signal_mask);
  sigaddset(&signal_mask, SIGPIPE);
wafwerar's avatar
wafwerar 已提交
881
  int32_t rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
882
  if (rc != 0) {
wafwerar's avatar
wafwerar 已提交
883
    // printf("failed to block SIGPIPE");
884
  }
885
#endif
886
}
887 888

uint32_t taosGetIpv4FromFqdn(const char *fqdn) {
wafwerar's avatar
wafwerar 已提交
889 890 891
#ifdef WINDOWS
  // Initialize Winsock
  WSADATA wsaData;
dengyihao's avatar
dengyihao 已提交
892
  int     iResult;
wafwerar's avatar
wafwerar 已提交
893 894
  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (iResult != 0) {
dengyihao's avatar
dengyihao 已提交
895 896
    // printf("WSAStartup failed: %d\n", iResult);
    return 1;
wafwerar's avatar
wafwerar 已提交
897 898
  }
#endif
899 900 901 902 903 904 905 906
  struct addrinfo hints = {0};
  hints.ai_family = AF_INET;
  hints.ai_socktype = SOCK_STREAM;

  struct addrinfo *result = NULL;

  int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);
  if (result) {
dengyihao's avatar
dengyihao 已提交
907
    struct sockaddr *   sa = result->ai_addr;
908 909 910 911 912 913 914 915
    struct sockaddr_in *si = (struct sockaddr_in *)sa;
    struct in_addr      ia = si->sin_addr;
    uint32_t            ip = ia.s_addr;
    freeaddrinfo(result);
    return ip;
  } else {
#ifdef EAI_SYSTEM
    if (ret == EAI_SYSTEM) {
dengyihao's avatar
dengyihao 已提交
916
      // printf("failed to get the ip address, fqdn:%s, errno:%d, since:%s", fqdn, errno, strerror(errno));
917
    } else {
dengyihao's avatar
dengyihao 已提交
918
      // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
919
    }
920
#else
dengyihao's avatar
dengyihao 已提交
921
    // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
922 923 924 925 926 927 928 929 930
#endif
    return 0xFFFFFFFF;
  }
}

int32_t taosGetFqdn(char *fqdn) {
  char hostname[1024];
  hostname[1023] = '\0';
  if (gethostname(hostname, 1023) == -1) {
dengyihao's avatar
dengyihao 已提交
931
    // printf("failed to get hostname, reason:%s", strerror(errno));
wafwerar's avatar
wafwerar 已提交
932
    assert(0);
933 934 935 936 937 938 939 940 941 942 943
    return -1;
  }

  struct addrinfo  hints = {0};
  struct addrinfo *result = NULL;
#ifdef __APPLE__
  // on macosx, hostname -f has the form of xxx.local
  // which will block getaddrinfo for a few seconds if AI_CANONNAME is set
  // thus, we choose AF_INET (ipv4 for the moment) to make getaddrinfo return
  // immediately
  hints.ai_family = AF_INET;
wafwerar's avatar
wafwerar 已提交
944
#else   // __APPLE__
945
  hints.ai_flags = AI_CANONNAME;
wafwerar's avatar
wafwerar 已提交
946
#endif  // __APPLE__
947 948
  int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
  if (!result) {
dengyihao's avatar
dengyihao 已提交
949
    fprintf(stderr, "failed to get fqdn, code:%d, reason:%s", ret, gai_strerror(ret));
950 951 952 953 954 955
    return -1;
  }

#ifdef __APPLE__
  // refer to comments above
  strcpy(fqdn, hostname);
wafwerar's avatar
wafwerar 已提交
956
#else   // __APPLE__
957
  strcpy(fqdn, result->ai_canonname);
wafwerar's avatar
wafwerar 已提交
958
#endif  // __APPLE__
959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994
  freeaddrinfo(result);
  return 0;
}

// Function converting an IP address string to an uint32_t.
uint32_t ip2uint(const char *const ip_addr) {
  char ip_addr_cpy[20];
  char ip[5];

  tstrncpy(ip_addr_cpy, ip_addr, sizeof(ip_addr_cpy));

  char *s_start, *s_end;
  s_start = ip_addr_cpy;
  s_end = ip_addr_cpy;

  int32_t k;

  for (k = 0; *s_start != '\0'; s_start = s_end) {
    for (s_end = s_start; *s_end != '.' && *s_end != '\0'; s_end++) {
    }
    if (*s_end == '.') {
      *s_end = '\0';
      s_end++;
    }
    ip[k++] = (char)atoi(s_start);
  }

  ip[k] = '\0';

  return *((uint32_t *)ip);
}

void tinet_ntoa(char *ipstr, uint32_t ip) {
  sprintf(ipstr, "%d.%d.%d.%d", ip & 0xFF, (ip >> 8) & 0xFF, (ip >> 16) & 0xFF, ip >> 24);
}

dengyihao's avatar
dengyihao 已提交
995
void taosIgnSIGPIPE() { signal(SIGPIPE, SIG_IGN); }
996 997

void taosSetMaskSIGPIPE() {
wafwerar's avatar
wafwerar 已提交
998
#ifdef WINDOWS
999
  // assert(0);
1000 1001 1002 1003
#else
  sigset_t signal_mask;
  sigemptyset(&signal_mask);
  sigaddset(&signal_mask, SIGPIPE);
wafwerar's avatar
wafwerar 已提交
1004
  int32_t rc = pthread_sigmask(SIG_SETMASK, &signal_mask, NULL);
1005
  if (rc != 0) {
wafwerar's avatar
wafwerar 已提交
1006 1007 1008 1009 1010
    // printf("failed to setmask SIGPIPE");
  }
#endif
}

wafwerar's avatar
wafwerar 已提交
1011
int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *addrLen) {
wafwerar's avatar
wafwerar 已提交
1012 1013 1014 1015 1016 1017 1018 1019
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
  return getsockname(pSocket->fd, destAddr, addrLen);
}

TdEpollPtr taosCreateEpoll(int32_t size) {
  EpollFd fd = -1;
wafwerar's avatar
wafwerar 已提交
1020
#ifdef WINDOWS
1021
  assert(0);
wafwerar's avatar
wafwerar 已提交
1022 1023 1024 1025 1026 1027 1028
#else
  fd = epoll_create(size);
#endif
  if (fd < 0) {
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
1029
  TdEpollPtr pEpoll = (TdEpollPtr)taosMemoryMalloc(sizeof(TdEpoll));
wafwerar's avatar
wafwerar 已提交
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042
  if (pEpoll == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }
  pEpoll->fd = fd;
  pEpoll->refId = 0;
  return pEpoll;
}
int32_t taosCtlEpoll(TdEpollPtr pEpoll, int32_t epollOperate, TdSocketPtr pSocket, struct epoll_event *event) {
  int32_t code = -1;
  if (pEpoll == NULL || pEpoll->fd < 0) {
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
1043
#ifdef WINDOWS
1044
  assert(0);
wafwerar's avatar
wafwerar 已提交
1045 1046 1047 1048 1049 1050 1051 1052 1053
#else
  code = epoll_ctl(pEpoll->fd, epollOperate, pSocket->fd, event);
#endif
  return code;
}
int32_t taosWaitEpoll(TdEpollPtr pEpoll, struct epoll_event *event, int32_t maxEvents, int32_t timeout) {
  int32_t code = -1;
  if (pEpoll == NULL || pEpoll->fd < 0) {
    return -1;
1054
  }
wafwerar's avatar
wafwerar 已提交
1055
#ifdef WINDOWS
1056
  assert(0);
wafwerar's avatar
wafwerar 已提交
1057 1058
#else
  code = epoll_wait(pEpoll->fd, event, maxEvents, timeout);
1059
#endif
wafwerar's avatar
wafwerar 已提交
1060 1061 1062 1063 1064 1065 1066 1067 1068
  return code;
}
int32_t taosCloseEpoll(TdEpollPtr *ppEpoll) {
  int32_t code;
  if (ppEpoll == NULL || *ppEpoll == NULL || (*ppEpoll)->fd < 0) {
    return -1;
  }
  code = taosCloseSocketNoCheck1((*ppEpoll)->fd);
  (*ppEpoll)->fd = -1;
wafwerar's avatar
wafwerar 已提交
1069
  taosMemoryFree(*ppEpoll);
wafwerar's avatar
wafwerar 已提交
1070 1071
  return code;
}
dengyihao's avatar
dengyihao 已提交
1072 1073 1074 1075
/*
 * Set TCP connection timeout per-socket level.
 * ref [https://github.com/libuv/help/issues/54]
 */
dengyihao's avatar
dengyihao 已提交
1076
int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
dengyihao's avatar
dengyihao 已提交
1077 1078 1079 1080 1081 1082 1083 1084 1085
#if defined(WINDOWS)
  SOCKET fd;
#else
  int      fd;
#endif
  if ((fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {
    return -1;
  }
#if defined(WINDOWS)
dengyihao's avatar
dengyihao 已提交
1086
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
dengyihao's avatar
dengyihao 已提交
1087 1088 1089
    return -1;
  }
#else  // Linux like systems
dengyihao's avatar
dengyihao 已提交
1090
  uint32_t conn_timeout_ms = timeout * 1000;
dengyihao's avatar
dengyihao 已提交
1091 1092 1093 1094 1095 1096
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
    return -1;
  }
#endif
  return (int)fd;
}