osSocket.c 27.9 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 54 55 56 57 58 59
typedef struct TdSocket {
#if SOCKET_WITH_LOCK
  TdThreadRwlock rwlock;
#endif
  int      refId;
  SocketFd fd;
} * TdSocketPtr, TdSocket;

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

wafwerar's avatar
wafwerar 已提交
68 69
typedef struct TdEpoll {
#if SOCKET_WITH_LOCK
wafwerar's avatar
wafwerar 已提交
70
  TdThreadRwlock rwlock;
wafwerar's avatar
wafwerar 已提交
71
#endif
dengyihao's avatar
dengyihao 已提交
72 73 74
  int     refId;
  EpollFd fd;
} * TdEpollPtr, TdEpoll;
S
Shengliang Guan 已提交
75

wafwerar's avatar
wafwerar 已提交
76
int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags, const struct sockaddr *dest_addr,
dengyihao's avatar
dengyihao 已提交
77
                   int addrlen) {
wafwerar's avatar
wafwerar 已提交
78 79
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
80
  }
wafwerar's avatar
wafwerar 已提交
81
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
82 83 84 85 86 87 88 89
  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 已提交
90
  }
wafwerar's avatar
wafwerar 已提交
91
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
92 93 94 95 96 97 98 99 100
  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 已提交
101
  }
wafwerar's avatar
wafwerar 已提交
102
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
103 104 105 106 107 108
  return recv(pSocket->fd, buf, len, 0);
  ;
#else
  return read(pSocket->fd, buf, len);
#endif
}
S
Shengliang Guan 已提交
109

dengyihao's avatar
dengyihao 已提交
110 111
int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr,
                           int *addrLen) {
wafwerar's avatar
wafwerar 已提交
112 113
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
114
  }
wafwerar's avatar
wafwerar 已提交
115 116 117
  return recvfrom(pSocket->fd, buf, len, flags, destAddr, addrLen);
}
int32_t taosCloseSocketNoCheck1(SocketFd fd) {
wafwerar's avatar
wafwerar 已提交
118
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
119 120 121
  return closesocket(fd);
#else
  return close(fd);
S
Shengliang Guan 已提交
122
#endif
wafwerar's avatar
wafwerar 已提交
123 124 125 126 127 128 129 130
}
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 已提交
131
  taosMemoryFree(*ppSocket);
wafwerar's avatar
wafwerar 已提交
132 133 134 135 136 137 138 139 140
  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 已提交
141
  taosMemoryFree(*ppSocketServer);
wafwerar's avatar
wafwerar 已提交
142 143
  return code;
}
S
Shengliang Guan 已提交
144

wafwerar's avatar
wafwerar 已提交
145 146 147 148
int32_t taosShutDownSocketRD(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
S
Shengliang Guan 已提交
149
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
150
  return closesocket(pSocket->fd);
S
Shengliang Guan 已提交
151
#elif __APPLE__
wafwerar's avatar
wafwerar 已提交
152
  return close(pSocket->fd);
S
Shengliang Guan 已提交
153
#else
wafwerar's avatar
wafwerar 已提交
154
  return shutdown(pSocket->fd, SHUT_RD);
S
Shengliang Guan 已提交
155 156
#endif
}
wafwerar's avatar
wafwerar 已提交
157 158 159 160
int32_t taosShutDownSocketServerRD(TdSocketServerPtr pSocketServer) {
  if (pSocketServer == NULL || pSocketServer->fd < 0) {
    return -1;
  }
S
Shengliang Guan 已提交
161
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
162
  return closesocket(pSocketServer->fd);
S
Shengliang Guan 已提交
163
#elif __APPLE__
wafwerar's avatar
wafwerar 已提交
164
  return close(pSocketServer->fd);
S
Shengliang Guan 已提交
165
#else
wafwerar's avatar
wafwerar 已提交
166
  return shutdown(pSocketServer->fd, SHUT_RD);
S
Shengliang Guan 已提交
167 168
#endif
}
S
Shengliang Guan 已提交
169

wafwerar's avatar
wafwerar 已提交
170 171 172
int32_t taosShutDownSocketWR(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
173
  }
wafwerar's avatar
wafwerar 已提交
174 175 176 177 178 179
#ifdef WINDOWS
  return closesocket(pSocket->fd);
#elif __APPLE__
  return close(pSocket->fd);
#else
  return shutdown(pSocket->fd, SHUT_WR);
S
slguan 已提交
180
#endif
S
Shengliang Guan 已提交
181
}
wafwerar's avatar
wafwerar 已提交
182 183 184 185 186 187 188 189 190 191
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 已提交
192
#endif
wafwerar's avatar
wafwerar 已提交
193 194 195 196
}
int32_t taosShutDownSocketRDWR(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
S
Shengliang Guan 已提交
197
  }
wafwerar's avatar
wafwerar 已提交
198 199 200 201 202 203 204 205 206 207 208
#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 已提交
209
  }
wafwerar's avatar
wafwerar 已提交
210 211 212 213 214 215 216
#ifdef WINDOWS
  return closesocket(pSocketServer->fd);
#elif __APPLE__
  return close(pSocketServer->fd);
#else
  return shutdown(pSocketServer->fd, SHUT_RDWR);
#endif
217
}
wafwerar's avatar
wafwerar 已提交
218

wafwerar's avatar
wafwerar 已提交
219 220
void taosWinSocketInit() {
#ifdef WINDOWS
S
Shengliang Guan 已提交
221 222 223 224 225 226 227 228 229
  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 已提交
230 231
#else
#endif
232
}
wafwerar's avatar
wafwerar 已提交
233 234 235 236
int32_t taosSetNonblocking(TdSocketPtr pSocket, int32_t on) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
237
#ifdef WINDOWS
S
Shengliang Guan 已提交
238 239 240
  u_long mode;
  if (on) {
    mode = 1;
wafwerar's avatar
wafwerar 已提交
241
    ioctlsocket(pSocket->fd, FIONBIO, &mode);
S
Shengliang Guan 已提交
242 243
  } else {
    mode = 0;
wafwerar's avatar
wafwerar 已提交
244 245 246 247 248 249 250
    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 已提交
251
  }
wafwerar's avatar
wafwerar 已提交
252 253 254 255 256 257 258 259 260 261 262

  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 已提交
263 264
  return 0;
}
wafwerar's avatar
wafwerar 已提交
265 266 267 268
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 已提交
269
#ifdef WINDOWS
S
Shengliang Guan 已提交
270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285
  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 已提交
286 287
  return setsockopt(pSocket->fd, level, optname, optval, optlen);
#else
wafwerar's avatar
wafwerar 已提交
288
  return setsockopt(pSocket->fd, level, optname, optval, (int)optlen);
wafwerar's avatar
wafwerar 已提交
289 290 291 292 293 294
#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 已提交
295
#ifdef WINDOWS
296
  assert(0);
wafwerar's avatar
wafwerar 已提交
297 298
  return 0;
#else
wafwerar's avatar
wafwerar 已提交
299
  return getsockopt(pSocket->fd, level, optname, optval, (int *)optlen);
wafwerar's avatar
wafwerar 已提交
300
#endif
S
Shengliang Guan 已提交
301
}
S
config  
Shengliang Guan 已提交
302
uint32_t taosInetAddr(const char *ipAddr) {
wafwerar's avatar
wafwerar 已提交
303
#ifdef WINDOWS
S
Shengliang Guan 已提交
304 305 306 307 308 309 310
  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 已提交
311 312 313
#else
  return inet_addr(ipAddr);
#endif
S
Shengliang Guan 已提交
314
}
S
Shengliang Guan 已提交
315
const char *taosInetNtoa(struct in_addr ipInt) {
wafwerar's avatar
wafwerar 已提交
316
#ifdef WINDOWS
S
Shengliang Guan 已提交
317 318 319
  // 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 已提交
320 321
#else
  return inet_ntoa(ipInt);
322
#endif
wafwerar's avatar
wafwerar 已提交
323
}
324 325

#ifndef SIGPIPE
wafwerar's avatar
wafwerar 已提交
326
#define SIGPIPE EPIPE
327 328 329 330
#endif

#define TCP_CONN_TIMEOUT 3000  // conn timeout

wafwerar's avatar
wafwerar 已提交
331 332 333 334
int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
335
  int32_t nleft, nwritten;
dengyihao's avatar
dengyihao 已提交
336
  char *  ptr = (char *)buf;
337 338 339 340

  nleft = nbytes;

  while (nleft > 0) {
wafwerar's avatar
wafwerar 已提交
341
    nwritten = taosWriteSocket(pSocket, (char *)ptr, (size_t)nleft);
342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359
    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 已提交
360 361 362 363
int32_t taosReadMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
364
  int32_t nleft, nread;
dengyihao's avatar
dengyihao 已提交
365
  char *  ptr = (char *)buf;
366 367 368 369

  nleft = nbytes;

  while (nleft > 0) {
wafwerar's avatar
wafwerar 已提交
370
    nread = taosReadSocket(pSocket, ptr, (size_t)nleft);
371 372 373
    if (nread == 0) {
      break;
    } else if (nread < 0) {
wafwerar's avatar
wafwerar 已提交
374
      if (errno == EINTR /* || errno == EAGAIN || errno == EWOULDBLOCK*/) {
375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391
        continue;
      } else {
        return -1;
      }
    } else {
      nleft -= nread;
      ptr += nread;
    }

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

  return (nbytes - nleft);
}

wafwerar's avatar
wafwerar 已提交
392 393 394 395 396
int32_t taosNonblockwrite(TdSocketPtr pSocket, char *ptr, int32_t nbytes) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
  taosSetNonblocking(pSocket, 1);
397

wafwerar's avatar
wafwerar 已提交
398 399
  int32_t        nleft, nwritten, nready;
  fd_set         fset;
400 401 402 403 404 405 406
  struct timeval tv;

  nleft = nbytes;
  while (nleft > 0) {
    tv.tv_sec = 30;
    tv.tv_usec = 0;
    FD_ZERO(&fset);
wafwerar's avatar
wafwerar 已提交
407 408
    FD_SET(pSocket->fd, &fset);
    if ((nready = select((SocketFd)(pSocket->fd + 1), NULL, &fset, NULL, &tv)) == 0) {
409
      errno = ETIMEDOUT;
wafwerar's avatar
wafwerar 已提交
410
      // printf("fd %d timeout, no enough space to write", fd);
411 412 413 414 415
      break;

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

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

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

wafwerar's avatar
wafwerar 已提交
424
      // printf("write error, %d (%s)", errno, strerror(errno));
425 426 427 428 429 430 431
      return -1;
    }

    nleft -= nwritten;
    ptr += nwritten;
  }

wafwerar's avatar
wafwerar 已提交
432
  taosSetNonblocking(pSocket, 0);
433 434 435 436

  return (nbytes - nleft);
}

wafwerar's avatar
wafwerar 已提交
437
TdSocketPtr taosOpenUdpSocket(uint32_t ip, uint16_t port) {
438
  struct sockaddr_in localAddr;
wafwerar's avatar
wafwerar 已提交
439 440
  SocketFd           fd;
  int32_t            bufSize = 1024000;
441

wafwerar's avatar
wafwerar 已提交
442
  // printf("open udp socket:0x%x:%hu", ip, port);
443 444 445 446 447 448

  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 已提交
449 450 451 452
  if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) <= 2) {
    // printf("failed to open udp socket: %d (%s)", errno, strerror(errno));
    taosCloseSocketNoCheck1(fd);
    return NULL;
453 454
  }

wafwerar's avatar
wafwerar 已提交
455
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
456 457 458
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
459
  }
wafwerar's avatar
wafwerar 已提交
460 461
  pSocket->fd = fd;
  pSocket->refId = 0;
462

wafwerar's avatar
wafwerar 已提交
463 464 465 466 467 468 469 470 471 472
  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;
473 474 475
  }

  /* bind socket to local address */
wafwerar's avatar
wafwerar 已提交
476 477 478 479
  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;
480 481
  }

wafwerar's avatar
wafwerar 已提交
482
  return pSocket;
483 484
}

wafwerar's avatar
wafwerar 已提交
485 486 487
TdSocketPtr taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientIp) {
  SocketFd           fd = -1;
  int32_t            ret;
488
  struct sockaddr_in serverAddr, clientAddr;
wafwerar's avatar
wafwerar 已提交
489
  int32_t            bufSize = 1024 * 1024;
490

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

wafwerar's avatar
wafwerar 已提交
493 494 495 496
  if (fd <= 2) {
    // printf("failed to open the socket: %d (%s)", errno, strerror(errno));
    if (fd >= 0) taosCloseSocketNoCheck1(fd);
    return NULL;
497 498
  }

wafwerar's avatar
wafwerar 已提交
499
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
500 501 502 503 504 505 506
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }
  pSocket->fd = fd;
  pSocket->refId = 0;

507 508
  /* set REUSEADDR option, so the portnumber can be re-used */
  int32_t reuse = 1;
wafwerar's avatar
wafwerar 已提交
509 510 511 512
  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;
513 514
  }

wafwerar's avatar
wafwerar 已提交
515 516 517 518
  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;
519 520
  }

wafwerar's avatar
wafwerar 已提交
521 522 523 524
  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;
525 526 527 528 529 530 531 532 533
  }

  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 已提交
534 535 536 537 538
    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;
539 540 541 542 543 544 545 546
    }
  }

  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 已提交
547 548 549
#ifdef _TD_LINUX
  taosSetNonblocking(pSocket, 1);
  ret = connect(pSocket->fd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
550 551
  if (ret == -1) {
    if (errno == EHOSTUNREACH) {
wafwerar's avatar
wafwerar 已提交
552 553 554
      // printf("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
      taosCloseSocket(&pSocket);
      return -1;
555
    } else if (errno == EINPROGRESS || errno == EAGAIN || errno == EWOULDBLOCK) {
wafwerar's avatar
wafwerar 已提交
556
      struct pollfd wfd[1];
557

wafwerar's avatar
wafwerar 已提交
558
      wfd[0].fd = pSocket->fd;
559
      wfd[0].events = POLLOUT;
wafwerar's avatar
wafwerar 已提交
560

561 562
      int res = poll(wfd, 1, TCP_CONN_TIMEOUT);
      if (res == -1 || res == 0) {
wafwerar's avatar
wafwerar 已提交
563 564
        // printf("failed to connect socket, ip:0x%x, port:%hu(poll error/conn timeout)", destIp, destPort);
        taosCloseSocket(&pSocket);  //
565 566
        return -1;
      }
wafwerar's avatar
wafwerar 已提交
567 568 569 570
      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);  //
571 572 573
        return -1;
      }
      ret = 0;
wafwerar's avatar
wafwerar 已提交
574 575 576 577 578
    } else {  // Other error
      // printf("failed to connect socket, ip:0x%x, port:%hu(target host cannot be reached)", destIp, destPort);
      taosCloseSocket(&pSocket);  //
      return -1;
    }
579
  }
wafwerar's avatar
wafwerar 已提交
580
  taosSetNonblocking(pSocket, 0);
581 582

#else
wafwerar's avatar
wafwerar 已提交
583
  ret = connect(pSocket->fd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
584 585 586
#endif

  if (ret != 0) {
wafwerar's avatar
wafwerar 已提交
587 588 589
    // printf("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
590
  } else {
wafwerar's avatar
wafwerar 已提交
591
    taosKeepTcpAlive(pSocket);
592 593
  }

wafwerar's avatar
wafwerar 已提交
594
  return pSocket;
595 596
}

wafwerar's avatar
wafwerar 已提交
597 598 599 600
int32_t taosKeepTcpAlive(TdSocketPtr pSocket) {
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
601
  int32_t alive = 1;
wafwerar's avatar
wafwerar 已提交
602 603 604
  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);
605 606 607 608 609 610
    return -1;
  }

#ifndef __APPLE__
  // all fails on macosx
  int32_t probes = 3;
wafwerar's avatar
wafwerar 已提交
611 612 613
  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);
614 615 616 617
    return -1;
  }

  int32_t alivetime = 10;
wafwerar's avatar
wafwerar 已提交
618 619 620
  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);
621 622 623 624
    return -1;
  }

  int32_t interval = 3;
wafwerar's avatar
wafwerar 已提交
625 626 627
  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);
628 629
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
630
#endif  // __APPLE__
631 632

  int32_t nodelay = 1;
wafwerar's avatar
wafwerar 已提交
633 634 635
  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);
636 637 638 639 640 641
    return -1;
  }

  struct linger linger = {0};
  linger.l_onoff = 1;
  linger.l_linger = 3;
wafwerar's avatar
wafwerar 已提交
642 643 644
  if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_LINGER, (void *)&linger, sizeof(linger)) < 0) {
    // printf("setsockopt SO_LINGER failed: %d (%s)", errno, strerror(errno));
    taosCloseSocket(&pSocket);
645 646 647 648 649 650
    return -1;
  }

  return 0;
}

dengyihao's avatar
dengyihao 已提交
651 652 653
int taosGetLocalIp(const char *eth, char *ip) {
#if defined(WINDOWS)
  // DO NOTHAING
654
  assert(0);
dengyihao's avatar
dengyihao 已提交
655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680
  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
681
  assert(0);
dengyihao's avatar
dengyihao 已提交
682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705
  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 已提交
706 707 708 709 710
      ret = -1;
      if (ip == (uint32_t)taosInetAddr(ip_str)) {
        ret = 0;
        break;
      }
dengyihao's avatar
dengyihao 已提交
711 712 713 714 715 716 717 718 719
      ifreq++;
    }
  }
  taosCloseSocketNoCheck1(fd);
  return ret;
#endif
  return 0;
}

720 721 722 723 724 725 726 727 728
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 已提交
729 730 731
#ifdef WINDOWS
  serverAdd.sin_addr.s_addr = INADDR_ANY;
#else
732
  serverAdd.sin_addr.s_addr = ip;
wafwerar's avatar
wafwerar 已提交
733
#endif
734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754
  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 已提交
755
    return false;
756 757 758 759 760 761 762 763
  }
  /* 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 已提交
764 765
  return true;
  // return 0 == taosValidIp(ip) ? true : false;
766
}
wafwerar's avatar
wafwerar 已提交
767
TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
768
  struct sockaddr_in serverAdd;
wafwerar's avatar
wafwerar 已提交
769
  SocketFd           fd;
770 771
  int32_t            reuse;

wafwerar's avatar
wafwerar 已提交
772
  // printf("open tcp server socket:0x%x:%hu", ip, port);
773 774 775 776 777 778

  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 已提交
779 780 781 782 783 784
  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 已提交
785
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
786 787 788
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
789
  }
wafwerar's avatar
wafwerar 已提交
790 791
  pSocket->refId = 0;
  pSocket->fd = fd;
792 793 794

  /* set REUSEADDR option, so the portnumber can be re-used */
  reuse = 1;
wafwerar's avatar
wafwerar 已提交
795 796 797 798
  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;
799 800 801
  }

  /* bind socket to server address */
wafwerar's avatar
wafwerar 已提交
802 803 804 805
  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;
806 807
  }

wafwerar's avatar
wafwerar 已提交
808 809 810 811
  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;
812 813
  }

wafwerar's avatar
wafwerar 已提交
814 815 816 817
  if (listen(pSocket->fd, 1024) < 0) {
    // printf("listen tcp server socket failed, 0x%x:%hu(%s)", ip, port, strerror(errno));
    taosCloseSocket(&pSocket);
    return NULL;
818 819
  }

wafwerar's avatar
wafwerar 已提交
820
  return (TdSocketServerPtr)pSocket;
821 822
}

dengyihao's avatar
dengyihao 已提交
823
TdSocketPtr taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr, int *addrLen) {
wafwerar's avatar
wafwerar 已提交
824 825 826 827 828 829 830 831 832
  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 已提交
833
  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
wafwerar's avatar
wafwerar 已提交
834 835 836 837 838 839 840 841
  if (pSocket == NULL) {
    taosCloseSocketNoCheck1(fd);
    return NULL;
  }
  pSocket->fd = fd;
  pSocket->refId = 0;
  return pSocket;
}
842 843 844
#define COPY_SIZE 32768
// sendfile shall be used

wafwerar's avatar
wafwerar 已提交
845 846 847 848
int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len) {
  if (pSrcSocket == NULL || pSrcSocket->fd < 0 || pDestSocket == NULL || pDestSocket->fd < 0) {
    return -1;
  }
849 850 851 852 853 854 855 856 857 858 859 860
  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 已提交
861
    int64_t retLen = taosReadMsg(pSrcSocket, temp, (int32_t)readLen);
862
    if (readLen != retLen) {
wafwerar's avatar
wafwerar 已提交
863 864
      // printf("read error, readLen:%" PRId64 " retLen:%" PRId64 " len:%" PRId64 " leftLen:%" PRId64 ", reason:%s",
      //        readLen, retLen, len, leftLen, strerror(errno));
865 866 867
      return -1;
    }

wafwerar's avatar
wafwerar 已提交
868
    writeLen = taosWriteMsg(pDestSocket, temp, (int32_t)readLen);
869 870

    if (readLen != writeLen) {
wafwerar's avatar
wafwerar 已提交
871 872
      // printf("copy error, readLen:%" PRId64 " writeLen:%" PRId64 " len:%" PRId64 " leftLen:%" PRId64 ", reason:%s",
      //        readLen, writeLen, len, leftLen, strerror(errno));
873 874 875 876 877 878 879 880
      return -1;
    }

    leftLen -= readLen;
  }

  return len;
}
881 882

void taosBlockSIGPIPE() {
wafwerar's avatar
wafwerar 已提交
883
#ifdef WINDOWS
884
  // assert(0);
885
#else
886 887 888
  sigset_t signal_mask;
  sigemptyset(&signal_mask);
  sigaddset(&signal_mask, SIGPIPE);
wafwerar's avatar
wafwerar 已提交
889
  int32_t rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
890
  if (rc != 0) {
wafwerar's avatar
wafwerar 已提交
891
    // printf("failed to block SIGPIPE");
892
  }
893
#endif
894
}
895 896

uint32_t taosGetIpv4FromFqdn(const char *fqdn) {
wafwerar's avatar
wafwerar 已提交
897 898 899
#ifdef WINDOWS
  // Initialize Winsock
  WSADATA wsaData;
dengyihao's avatar
dengyihao 已提交
900
  int     iResult;
wafwerar's avatar
wafwerar 已提交
901 902
  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (iResult != 0) {
dengyihao's avatar
dengyihao 已提交
903 904
    // printf("WSAStartup failed: %d\n", iResult);
    return 1;
wafwerar's avatar
wafwerar 已提交
905 906
  }
#endif
907 908 909 910 911 912 913 914
  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 已提交
915
    struct sockaddr *   sa = result->ai_addr;
916 917 918 919 920 921 922 923
    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 已提交
924
      // printf("failed to get the ip address, fqdn:%s, errno:%d, since:%s", fqdn, errno, strerror(errno));
925
    } else {
dengyihao's avatar
dengyihao 已提交
926
      // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
927
    }
928
#else
dengyihao's avatar
dengyihao 已提交
929
    // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
930 931 932 933 934 935
#endif
    return 0xFFFFFFFF;
  }
}

int32_t taosGetFqdn(char *fqdn) {
wafwerar's avatar
wafwerar 已提交
936 937 938 939 940 941 942 943 944 945
#ifdef WINDOWS
  // Initialize Winsock
  WSADATA wsaData;
  int     iResult;
  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (iResult != 0) {
    // printf("WSAStartup failed: %d\n", iResult);
    return 1;
  }
#endif
946 947 948
  char hostname[1024];
  hostname[1023] = '\0';
  if (gethostname(hostname, 1023) == -1) {
wafwerar's avatar
wafwerar 已提交
949
#ifdef WINDOWS
wafwerar's avatar
wafwerar 已提交
950
    printf("failed to get hostname, reason:%s\n", strerror(WSAGetLastError()));
wafwerar's avatar
wafwerar 已提交
951
#else
wafwerar's avatar
wafwerar 已提交
952
    printf("failed to get hostname, reason:%s\n", strerror(errno));
wafwerar's avatar
wafwerar 已提交
953
#endif
wafwerar's avatar
wafwerar 已提交
954
    assert(0);
955 956 957 958 959 960 961 962 963 964 965
    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 已提交
966
#else   // __APPLE__
967
  hints.ai_flags = AI_CANONNAME;
wafwerar's avatar
wafwerar 已提交
968
#endif  // __APPLE__
969 970
  int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
  if (!result) {
wafwerar's avatar
wafwerar 已提交
971
    fprintf(stderr, "failed to get fqdn, code:%d, reason:%s\n", ret, gai_strerror(ret));
972 973 974 975 976 977
    return -1;
  }

#ifdef __APPLE__
  // refer to comments above
  strcpy(fqdn, hostname);
wafwerar's avatar
wafwerar 已提交
978
#else   // __APPLE__
979
  strcpy(fqdn, result->ai_canonname);
wafwerar's avatar
wafwerar 已提交
980
#endif  // __APPLE__
981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016
  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 已提交
1017
void taosIgnSIGPIPE() { signal(SIGPIPE, SIG_IGN); }
1018 1019

void taosSetMaskSIGPIPE() {
wafwerar's avatar
wafwerar 已提交
1020
#ifdef WINDOWS
1021
  // assert(0);
1022 1023 1024 1025
#else
  sigset_t signal_mask;
  sigemptyset(&signal_mask);
  sigaddset(&signal_mask, SIGPIPE);
wafwerar's avatar
wafwerar 已提交
1026
  int32_t rc = pthread_sigmask(SIG_SETMASK, &signal_mask, NULL);
1027
  if (rc != 0) {
wafwerar's avatar
wafwerar 已提交
1028 1029 1030 1031 1032
    // printf("failed to setmask SIGPIPE");
  }
#endif
}

wafwerar's avatar
wafwerar 已提交
1033
int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *addrLen) {
wafwerar's avatar
wafwerar 已提交
1034 1035 1036 1037 1038 1039
  if (pSocket == NULL || pSocket->fd < 0) {
    return -1;
  }
  return getsockname(pSocket->fd, destAddr, addrLen);
}

dengyihao's avatar
dengyihao 已提交
1040 1041 1042 1043
/*
 * Set TCP connection timeout per-socket level.
 * ref [https://github.com/libuv/help/issues/54]
 */
dengyihao's avatar
dengyihao 已提交
1044
int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
dengyihao's avatar
dengyihao 已提交
1045 1046 1047 1048 1049 1050 1051 1052 1053
#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 已提交
1054
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
dengyihao's avatar
dengyihao 已提交
1055 1056
    return -1;
  }
wafwerar's avatar
wafwerar 已提交
1057 1058 1059 1060 1061
#elif defined(_TD_DARWIN_64)
  uint32_t conn_timeout_ms = timeout * 1000;
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_CONNECTIONTIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
    return -1;
  }
dengyihao's avatar
dengyihao 已提交
1062
#else  // Linux like systems
dengyihao's avatar
dengyihao 已提交
1063
  uint32_t conn_timeout_ms = timeout * 1000;
dengyihao's avatar
dengyihao 已提交
1064 1065 1066 1067 1068 1069
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
    return -1;
  }
#endif
  return (int)fd;
}