#3522

Committed 07 Nov 2024 05:59AM UTC coverage: 58.216% (+1.3%) from 56.943%

Build # #3522

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28663 from taosdata/fix/3_liaohj

fix(stream): stop the underlying scan operations for stream

Run Details

111884 of 248391 branches covered (45.04%)

Branch coverage included in aggregate %.

3 of 4 new or added lines in 1 file covered. (75.0%)

1164 existing lines in 134 files now uncovered.

191720 of 273118 relevant lines covered (70.2%)

13088725.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.57

/source/os/src/osSocket.c

/*
 * 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
#define ALLOW_FORBID_FUNC
#include "os.h"

#if defined(WINDOWS)
#include <IPHlpApi.h>
#include <WS2tcpip.h>
#include <Winsock2.h>
#include <stdio.h>
#include <string.h>
#include <tchar.h>
#include <winbase.h>
#else
#include <arpa/inet.h>
#include <fcntl.h>
#include <net/if.h>
#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>

#if defined(DARWIN)
#include <dispatch/dispatch.h>
#include "osEok.h"
#else
#include <sys/epoll.h>
#endif
#endif

#ifndef INVALID_SOCKET
#define INVALID_SOCKET -1
#endif

typedef struct TdSocket {
#if SOCKET_WITH_LOCK
  TdThreadRwlock rwlock;
#endif
  int      refId;
  SocketFd fd;
} * TdSocketPtr, TdSocket;

typedef struct TdSocketServer {
#if SOCKET_WITH_LOCK
  TdThreadRwlock rwlock;
#endif
  int      refId;
  SocketFd fd;
} * TdSocketServerPtr, TdSocketServer;

typedef struct TdEpoll {
#if SOCKET_WITH_LOCK
  TdThreadRwlock rwlock;
#endif
  int     refId;
  EpollFd fd;
} * TdEpollPtr, TdEpoll;

int32_t taosCloseSocketNoCheck1(SocketFd fd) {
#ifdef WINDOWS
  int ret = closesocket(fd);
  if (ret == SOCKET_ERROR) {
    int errorCode = WSAGetLastError();
    return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
  }
  return 0;
#else
  int32_t code = close(fd);
  if (-1 == code) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return terrno;
  }
  return code;
#endif
}

int32_t taosCloseSocket(TdSocketPtr *ppSocket) {
  int32_t code;
  if (ppSocket == NULL || *ppSocket == NULL || (*ppSocket)->fd < 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
  code = taosCloseSocketNoCheck1((*ppSocket)->fd);
  (*ppSocket)->fd = -1;
  taosMemoryFree(*ppSocket);

  return code;
}

int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t optlen) {
  if (pSocket == NULL || pSocket->fd < 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

#ifdef WINDOWS
#ifdef TCP_KEEPCNT
  if (level == SOL_SOCKET && optname == TCP_KEEPCNT) {
    return 0;
  }
#endif

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

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

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

  int ret = setsockopt(pSocket->fd, level, optname, optval, optlen);
  if (ret == SOCKET_ERROR) {
    int errorCode = WSAGetLastError();
    return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
  }
#else
  int32_t code = setsockopt(pSocket->fd, level, optname, optval, (int)optlen);
  if (-1 == code) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return terrno;
  }
  return 0;
#endif
}

const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len) {
  const char *r = inet_ntop(AF_INET, &ipInt, dstStr, len);
  if (NULL == r) {
    terrno = TAOS_SYSTEM_ERROR(errno);
  }

  return r;
}

#ifndef SIGPIPE
#define SIGPIPE EPIPE
#endif

#define TCP_CONN_TIMEOUT 3000  // conn timeout

bool taosValidIpAndPort(uint32_t ip, uint16_t port) {
  struct sockaddr_in serverAdd;
  SocketFd           fd;
  int32_t            reuse;
  int32_t            code = 0;

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

  bzero((char *)&serverAdd, sizeof(serverAdd));
  serverAdd.sin_family = AF_INET;
#ifdef WINDOWS
  serverAdd.sin_addr.s_addr = INADDR_ANY;
#else
  serverAdd.sin_addr.s_addr = ip;
#endif
  serverAdd.sin_port = (uint16_t)htons(port);

  fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  if (-1 == fd) {  // exception
    terrno = TAOS_SYSTEM_ERROR(errno);
    return false;
  }

  TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));
  if (pSocket == NULL) {
    TAOS_SKIP_ERROR(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) {
    TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));
    return false;
  }

  /* bind socket to server address */
  if (-1 == bind(pSocket->fd, (struct sockaddr *)&serverAdd, sizeof(serverAdd))) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));
    return false;
  }

  TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));

  return true;
}

int32_t taosBlockSIGPIPE() {
#ifdef WINDOWS
  return 0;
#else
  sigset_t signal_mask;
  (void)sigemptyset(&signal_mask);
  (void)sigaddset(&signal_mask, SIGPIPE);
  int32_t rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
  if (rc != 0) {
    terrno = TAOS_SYSTEM_ERROR(rc);
    return terrno;
  }

  return 0;
#endif
}

int32_t taosGetIpv4FromFqdn(const char *fqdn, uint32_t *ip) {
#ifdef WINDOWS
  // Initialize Winsock
  WSADATA wsaData;
  int     iResult;
  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (iResult != 0) {
    return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
  }
#endif

#if defined(LINUX)
  struct addrinfo hints = {0};
  hints.ai_family = AF_INET;
  hints.ai_socktype = SOCK_STREAM;

  struct addrinfo *result = NULL;
  bool             inRetry = false;

  while (true) {
    int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);
    if (ret) {
      if (EAI_AGAIN == ret && !inRetry) {
        inRetry = true;
        continue;
      } else if (EAI_SYSTEM == ret) {
        terrno = TAOS_SYSTEM_ERROR(errno);
        return terrno;
      }

      terrno = TAOS_SYSTEM_ERROR(errno);
      return terrno;
    }

    struct sockaddr    *sa = result->ai_addr;
    struct sockaddr_in *si = (struct sockaddr_in *)sa;
    struct in_addr      ia = si->sin_addr;

    *ip = ia.s_addr;

    freeaddrinfo(result);

    return 0;
  }
#else
  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) {
    struct sockaddr    *sa = result->ai_addr;
    struct sockaddr_in *si = (struct sockaddr_in *)sa;
    struct in_addr      ia = si->sin_addr;
    *ip = ia.s_addr;
    freeaddrinfo(result);
    return 0;
  } else {
#ifdef EAI_SYSTEM
    if (ret == EAI_SYSTEM) {
      // printf("failed to get the ip address, fqdn:%s, errno:%d, since:%s", fqdn, errno, strerror(errno));
    } else {
      // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
    }
#else
    // printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
#endif

    *ip = 0xFFFFFFFF;
    return TSDB_CODE_RPC_FQDN_ERROR;
  }
#endif
}

int32_t taosGetFqdn(char *fqdn) {
#ifdef WINDOWS
  // Initialize Winsock
  WSADATA wsaData;
  int     iResult;
  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (iResult != 0) {
    // printf("WSAStartup failed: %d\n", iResult);
    return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
  }
#endif
  char hostname[1024];
  hostname[1023] = '\0';
  int32_t code = taosGetlocalhostname(hostname, 1023);
  if (code) {
    return code;
  }

#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;
  tstrncpy(fqdn, hostname, TD_FQDN_LEN);
  tstrncpy(fqdn + strlen(hostname), ".local", TD_FQDN_LEN - strlen(hostname));
#else  // linux

#endif  // linux

#if defined(LINUX)

  struct addrinfo  hints = {0};
  struct addrinfo *result = NULL;
  hints.ai_flags = AI_CANONNAME;

  while (true) {
    int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
    if (ret) {
      if (EAI_AGAIN == ret) {
        continue;
      } else if (EAI_SYSTEM == ret) {
        terrno = TAOS_SYSTEM_ERROR(errno);
        return terrno;
      }

      terrno = TAOS_SYSTEM_ERROR(ret);
      return terrno;
    }

    break;
  }

  tstrncpy(fqdn, result->ai_canonname, TD_FQDN_LEN);

  freeaddrinfo(result);

#elif WINDOWS
  struct addrinfo  hints = {0};
  struct addrinfo *result = NULL;
  hints.ai_flags = AI_CANONNAME;

  int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
  if (!result) {
    // fprintf(stderr, "failed to get fqdn, code:%d, hostname:%s, reason:%s\n", ret, hostname, gai_strerror(ret));
    return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
  }
  tstrncpy(fqdn, result->ai_canonname, TD_FQDN_LEN);
  freeaddrinfo(result);

#endif

  return 0;
}

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

int32_t taosIgnSIGPIPE() {
  sighandler_t h = signal(SIGPIPE, SIG_IGN);
  if (SIG_ERR == h) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return terrno;
  }

  return 0;
}

/*
 * Set TCP connection timeout per-socket level.
 * ref [https://github.com/libuv/help/issues/54]
 */
int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
#if defined(WINDOWS)
  SOCKET fd;
#else
  int fd;
#endif

  if ((fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return terrno;
  }

#if defined(WINDOWS)
  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
    taosCloseSocketNoCheck1(fd);
    return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
  }
#elif defined(_TD_DARWIN_64)
  // invalid config
  // uint32_t conn_timeout_ms = timeout * 1000;
  // if (0 != setsockopt(fd, IPPROTO_TCP, TCP_CONNECTIONTIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
  //  taosCloseSocketNoCheck1(fd);
  //  return -1;
  //}
#else  // Linux like systems
  uint32_t conn_timeout_ms = timeout;
  if (-1 == setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    TAOS_SKIP_ERROR(taosCloseSocketNoCheck1(fd));
    return terrno;
  }
#endif

  return (int)fd;
}

int32_t taosWinSocketInit() {
#ifdef WINDOWS
  static int8_t flag = 0;
  if (atomic_val_compare_exchange_8(&flag, 0, 1) == 0) {
    WORD    wVersionRequested;
    WSADATA wsaData;
    wVersionRequested = MAKEWORD(1, 1);
    if (WSAStartup(wVersionRequested, &wsaData) != 0) {
      atomic_store_8(&flag, 0);
      int errorCode = WSAGetLastError();
      return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
    }
  }
  return 0;
#else
#endif
  return 0;
}

uint64_t taosHton64(uint64_t val) {
#if defined(WINDOWS) || defined(DARWIN)
  return ((val & 0x00000000000000ff) << 7 * 8) | ((val & 0x000000000000ff00) << 5 * 8) |
         ((val & 0x0000000000ff0000) << 3 * 8) | ((val & 0x00000000ff000000) << 1 * 8) |
         ((val & 0x000000ff00000000) >> 1 * 8) | ((val & 0x0000ff0000000000) >> 3 * 8) |
         ((val & 0x00ff000000000000) >> 5 * 8) | ((val & 0xff00000000000000) >> 7 * 8);
#else
  if (__BYTE_ORDER == __LITTLE_ENDIAN) {
    return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) | (unsigned int)htonl((int)(val >> 32));
  } else if (__BYTE_ORDER == __BIG_ENDIAN) {
    return val;
  }
#endif
}

uint64_t taosNtoh64(uint64_t val) {
#if defined(WINDOWS) || defined(DARWIN)
  return taosHton64(val);
#else
  if (__BYTE_ORDER == __LITTLE_ENDIAN) {
    return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) | (unsigned int)htonl((int)(val >> 32));
  } else if (__BYTE_ORDER == __BIG_ENDIAN) {
    return val;
  }
#endif
}

int32_t taosSetSockOpt2(int32_t fd) {
#if defined(WINDOWS) || defined(DARWIN)
  return 0;
#else
  int32_t ret = setsockopt(fd, IPPROTO_TCP, TCP_QUICKACK, (int[]){1}, sizeof(int));
  if (ret < 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    return terrno;
  } else {
    return 0;
  }
#endif
  return 0;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	#define ALLOW_FORBID_FUNC
18	#include "os.h"
19
20	#if defined(WINDOWS)
21	#include <IPHlpApi.h>
22	#include <WS2tcpip.h>
23	#include <Winsock2.h>
24	#include <stdio.h>
25	#include <string.h>
26	#include <tchar.h>
27	#include <winbase.h>
28	#else
29	#include <arpa/inet.h>
30	#include <fcntl.h>
31	#include <net/if.h>
32	#include <netdb.h>
33	#include <netinet/in.h>
34	#include <netinet/ip.h>
35	#include <netinet/tcp.h>
36	#include <netinet/udp.h>
37	#include <sys/socket.h>
38	#include <unistd.h>
39
40	#if defined(DARWIN)
41	#include <dispatch/dispatch.h>
42	#include "osEok.h"
43	#else
44	#include <sys/epoll.h>
45	#endif
46	#endif
47
48	#ifndef INVALID_SOCKET
49	#define INVALID_SOCKET -1
50	#endif
51
52	typedef struct TdSocket {
53	#if SOCKET_WITH_LOCK
54	TdThreadRwlock rwlock;
55	#endif
56	int refId;
57	SocketFd fd;
58	} * TdSocketPtr, TdSocket;
59
60	typedef struct TdSocketServer {
61	#if SOCKET_WITH_LOCK
62	TdThreadRwlock rwlock;
63	#endif
64	int refId;
65	SocketFd fd;
66	} * TdSocketServerPtr, TdSocketServer;
67
68	typedef struct TdEpoll {
69	#if SOCKET_WITH_LOCK
70	TdThreadRwlock rwlock;
71	#endif
72	int refId;
73	EpollFd fd;
74	} * TdEpollPtr, TdEpoll;
75
76	int32_t taosCloseSocketNoCheck1(SocketFd fd) {	1,166✔
77	#ifdef WINDOWS
78	int ret = closesocket(fd);
79	if (ret == SOCKET_ERROR) {
80	int errorCode = WSAGetLastError();
81	return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
82	}
83	return 0;
84	#else
85	int32_t code = close(fd);	1,166✔
86	if (-1 == code) {	1,166!
87	terrno = TAOS_SYSTEM_ERROR(errno);	×
88	return terrno;	×
89	}
90	return code;	1,166✔
91	#endif
92	}
93
94	int32_t taosCloseSocket(TdSocketPtr *ppSocket) {	1,166✔
95	int32_t code;
96	if (ppSocket == NULL \|\| ppSocket == NULL \|\| (ppSocket)->fd < 0) {	1,166!
97	terrno = TSDB_CODE_INVALID_PARA;	×
98	return terrno;	×
99	}
100	code = taosCloseSocketNoCheck1((*ppSocket)->fd);	1,166✔
101	(*ppSocket)->fd = -1;	1,166✔
102	taosMemoryFree(*ppSocket);	1,166✔
103
104	return code;	1,166✔
105	}
106
107	int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t optlen) {	1,166✔
108	if (pSocket == NULL \|\| pSocket->fd < 0) {	1,166!
109	terrno = TSDB_CODE_INVALID_PARA;	×
110	return terrno;	×
111	}
112
113	#ifdef WINDOWS
114	#ifdef TCP_KEEPCNT
115	if (level == SOL_SOCKET && optname == TCP_KEEPCNT) {
116	return 0;
117	}
118	#endif
119
120	#ifdef TCP_KEEPIDLE
121	if (level == SOL_TCP && optname == TCP_KEEPIDLE) {
122	return 0;
123	}
124	#endif
125
126	#ifdef TCP_KEEPINTVL
127	if (level == SOL_TCP && optname == TCP_KEEPINTVL) {
128	return 0;
129	}
130	#endif
131
132	#ifdef TCP_KEEPCNT
133	if (level == SOL_TCP && optname == TCP_KEEPCNT) {
134	return 0;
135	}
136	#endif
137
138	int ret = setsockopt(pSocket->fd, level, optname, optval, optlen);
139	if (ret == SOCKET_ERROR) {
140	int errorCode = WSAGetLastError();
141	return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
142	}
143	#else
144	int32_t code = setsockopt(pSocket->fd, level, optname, optval, (int)optlen);	1,166✔
145	if (-1 == code) {	1,166!
146	terrno = TAOS_SYSTEM_ERROR(errno);	×
147	return terrno;	×
148	}
149	return 0;	1,166✔
150	#endif
151	}
152
153	const char taosInetNtoa(struct in_addr ipInt, char dstStr, int32_t len) {	×
154	const char *r = inet_ntop(AF_INET, &ipInt, dstStr, len);	×
155	if (NULL == r) {	×
156	terrno = TAOS_SYSTEM_ERROR(errno);	×
157	}
158
159	return r;	×
160	}
161
162	#ifndef SIGPIPE
163	#define SIGPIPE EPIPE
164	#endif
165
166	#define TCP_CONN_TIMEOUT 3000 // conn timeout
167
168	bool taosValidIpAndPort(uint32_t ip, uint16_t port) {	1,166✔
169	struct sockaddr_in serverAdd;
170	SocketFd fd;
171	int32_t reuse;
172	int32_t code = 0;	1,166✔
173
174	// printf("open tcp server socket:0x%x:%hu", ip, port);
175
176	bzero((char *)&serverAdd, sizeof(serverAdd));	1,166✔
177	serverAdd.sin_family = AF_INET;	1,166✔
178	#ifdef WINDOWS
179	serverAdd.sin_addr.s_addr = INADDR_ANY;
180	#else
181	serverAdd.sin_addr.s_addr = ip;	1,166✔
182	#endif
183	serverAdd.sin_port = (uint16_t)htons(port);	1,166✔
184
185	fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);	1,166✔
186	if (-1 == fd) { // exception	1,166!
187	terrno = TAOS_SYSTEM_ERROR(errno);	×
188	return false;	×
189	}
190
191	TdSocketPtr pSocket = (TdSocketPtr)taosMemoryMalloc(sizeof(TdSocket));	1,166✔
192	if (pSocket == NULL) {	1,166!
193	TAOS_SKIP_ERROR(taosCloseSocketNoCheck1(fd));	×
194	return false;	×
195	}
196	pSocket->refId = 0;	1,166✔
197	pSocket->fd = fd;	1,166✔
198
199	/* set REUSEADDR option, so the portnumber can be re-used */
200	reuse = 1;	1,166✔
201	if (taosSetSockOpt(pSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {	1,166!
202	TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));	×
203	return false;	×
204	}
205
206	/* bind socket to server address */
207	if (-1 == bind(pSocket->fd, (struct sockaddr *)&serverAdd, sizeof(serverAdd))) {	1,166!
208	terrno = TAOS_SYSTEM_ERROR(errno);	×
209	TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));	×
210	return false;	×
211	}
212
213	TAOS_SKIP_ERROR(taosCloseSocket(&pSocket));	1,166✔
214
215	return true;	1,166✔
216	}
217
218	int32_t taosBlockSIGPIPE() {	514,723✔
219	#ifdef WINDOWS
220	return 0;
221	#else
222	sigset_t signal_mask;
223	(void)sigemptyset(&signal_mask);	514,723✔
224	(void)sigaddset(&signal_mask, SIGPIPE);	514,757✔
225	int32_t rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);	514,768✔
226	if (rc != 0) {	515,030!
227	terrno = TAOS_SYSTEM_ERROR(rc);	×
228	return terrno;	×
229	}
230
231	return 0;	515,030✔
232	#endif
233	}
234
235	int32_t taosGetIpv4FromFqdn(const char fqdn, uint32_t ip) {	101,700✔
236	#ifdef WINDOWS
237	// Initialize Winsock
238	WSADATA wsaData;
239	int iResult;
240	iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
241	if (iResult != 0) {
242	return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
243	}
244	#endif
245
246	#if defined(LINUX)
247	struct addrinfo hints = {0};	101,700✔
248	hints.ai_family = AF_INET;	101,700✔
249	hints.ai_socktype = SOCK_STREAM;	101,700✔
250
251	struct addrinfo *result = NULL;	101,700✔
252	bool inRetry = false;	101,700✔
253
254	while (true) {	×
255	int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);	101,700✔
256	if (ret) {	101,760✔
257	if (EAI_AGAIN == ret && !inRetry) {	9!
258	inRetry = true;	×
259	continue;	×
260	} else if (EAI_SYSTEM == ret) {	9!
261	terrno = TAOS_SYSTEM_ERROR(errno);	×
262	return terrno;	101,761✔
263	}
264
265	terrno = TAOS_SYSTEM_ERROR(errno);	9✔
266	return terrno;	9✔
267	}
268
269	struct sockaddr *sa = result->ai_addr;	101,751✔
270	struct sockaddr_in si = (struct sockaddr_in )sa;	101,751✔
271	struct in_addr ia = si->sin_addr;	101,751✔
272
273	*ip = ia.s_addr;	101,751✔
274
275	freeaddrinfo(result);	101,751✔
276
277	return 0;	101,749✔
278	}
279	#else
280	struct addrinfo hints = {0};
281	hints.ai_family = AF_INET;
282	hints.ai_socktype = SOCK_STREAM;
283
284	struct addrinfo *result = NULL;
285
286	int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);
287	if (result) {
288	struct sockaddr *sa = result->ai_addr;
289	struct sockaddr_in si = (struct sockaddr_in )sa;
290	struct in_addr ia = si->sin_addr;
291	*ip = ia.s_addr;
292	freeaddrinfo(result);
293	return 0;
294	} else {
295	#ifdef EAI_SYSTEM
296	if (ret == EAI_SYSTEM) {
297	// printf("failed to get the ip address, fqdn:%s, errno:%d, since:%s", fqdn, errno, strerror(errno));
298	} else {
299	// printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
300	}
301	#else
302	// printf("failed to get the ip address, fqdn:%s, ret:%d, since:%s", fqdn, ret, gai_strerror(ret));
303	#endif
304
305	*ip = 0xFFFFFFFF;
306	return TSDB_CODE_RPC_FQDN_ERROR;
307	}
308	#endif
309	}
310
311	int32_t taosGetFqdn(char *fqdn) {	39,645✔
312	#ifdef WINDOWS
313	// Initialize Winsock
314	WSADATA wsaData;
315	int iResult;
316	iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
317	if (iResult != 0) {
318	// printf("WSAStartup failed: %d\n", iResult);
319	return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
320	}
321	#endif
322	char hostname[1024];
323	hostname[1023] = '\0';	39,645✔
324	int32_t code = taosGetlocalhostname(hostname, 1023);	39,645✔
325	if (code) {	39,645!
326	return code;	×
327	}
328
329	#ifdef __APPLE__
330	// on macosx, hostname -f has the form of xxx.local
331	// which will block getaddrinfo for a few seconds if AI_CANONNAME is set
332	// thus, we choose AF_INET (ipv4 for the moment) to make getaddrinfo return
333	// immediately
334	// hints.ai_family = AF_INET;
335	tstrncpy(fqdn, hostname, TD_FQDN_LEN);
336	tstrncpy(fqdn + strlen(hostname), ".local", TD_FQDN_LEN - strlen(hostname));
337	#else // linux
338
339	#endif // linux
340
341	#if defined(LINUX)
342
343	struct addrinfo hints = {0};	39,645✔
344	struct addrinfo *result = NULL;	39,645✔
345	hints.ai_flags = AI_CANONNAME;	39,645✔
346
347	while (true) {	×
348	int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);	39,645✔
349	if (ret) {	39,645!
350	if (EAI_AGAIN == ret) {	×
351	continue;	×
352	} else if (EAI_SYSTEM == ret) {	×
353	terrno = TAOS_SYSTEM_ERROR(errno);	×
354	return terrno;	×
355	}
356
357	terrno = TAOS_SYSTEM_ERROR(ret);	×
358	return terrno;	×
359	}
360
361	break;	39,645✔
362	}
363
364	tstrncpy(fqdn, result->ai_canonname, TD_FQDN_LEN);	39,645✔
365
366	freeaddrinfo(result);	39,645✔
367
368	#elif WINDOWS
369	struct addrinfo hints = {0};
370	struct addrinfo *result = NULL;
371	hints.ai_flags = AI_CANONNAME;
372
373	int32_t ret = getaddrinfo(hostname, NULL, &hints, &result);
374	if (!result) {
375	// fprintf(stderr, "failed to get fqdn, code:%d, hostname:%s, reason:%s\n", ret, hostname, gai_strerror(ret));
376	return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
377	}
378	tstrncpy(fqdn, result->ai_canonname, TD_FQDN_LEN);
379	freeaddrinfo(result);
380
381	#endif
382
383	return 0;	39,645✔
384	}
385
386	void tinet_ntoa(char *ipstr, uint32_t ip) {	11,095✔
387	(void)snprintf(ipstr, TD_IP_LEN, "%d.%d.%d.%d", ip & 0xFF, (ip >> 8) & 0xFF, (ip >> 16) & 0xFF, ip >> 24);	11,095✔
388	}	11,095✔
389
390	int32_t taosIgnSIGPIPE() {	1,168✔
391	sighandler_t h = signal(SIGPIPE, SIG_IGN);	1,168✔
392	if (SIG_ERR == h) {	1,168!
393	terrno = TAOS_SYSTEM_ERROR(errno);	×
394	return terrno;	×
395	}
396
397	return 0;	1,168✔
398	}
399
400	/*
401	* Set TCP connection timeout per-socket level.
402	* ref [https://github.com/libuv/help/issues/54]
403	*/
404	int32_t taosCreateSocketWithTimeout(uint32_t timeout) {	129,446✔
405	#if defined(WINDOWS)
406	SOCKET fd;
407	#else
408	int fd;
409	#endif
410
411	if ((fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {	129,446!
412	terrno = TAOS_SYSTEM_ERROR(errno);	×
413	return terrno;	×
414	}
415
416	#if defined(WINDOWS)
417	if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
418	taosCloseSocketNoCheck1(fd);
419	return TAOS_SYSTEM_WINSOCKET_ERROR(WSAGetLastError());
420	}
421	#elif defined(_TD_DARWIN_64)
422	// invalid config
423	// uint32_t conn_timeout_ms = timeout * 1000;
424	// if (0 != setsockopt(fd, IPPROTO_TCP, TCP_CONNECTIONTIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
425	// taosCloseSocketNoCheck1(fd);
426	// return -1;
427	//}
428	#else // Linux like systems
429	uint32_t conn_timeout_ms = timeout;	129,446✔
430	if (-1 == setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {	129,446!
UNCOV 431	terrno = TAOS_SYSTEM_ERROR(errno);	×
432	TAOS_SKIP_ERROR(taosCloseSocketNoCheck1(fd));	×
433	return terrno;	×
434	}
435	#endif
436
437	return (int)fd;	129,439✔
438	}
439
440	int32_t taosWinSocketInit() {	×
441	#ifdef WINDOWS
442	static int8_t flag = 0;
443	if (atomic_val_compare_exchange_8(&flag, 0, 1) == 0) {
444	WORD wVersionRequested;
445	WSADATA wsaData;
446	wVersionRequested = MAKEWORD(1, 1);
447	if (WSAStartup(wVersionRequested, &wsaData) != 0) {
448	atomic_store_8(&flag, 0);
449	int errorCode = WSAGetLastError();
450	return terrno = TAOS_SYSTEM_WINSOCKET_ERROR(errorCode);
451	}
452	}
453	return 0;
454	#else
455	#endif
456	return 0;	×
457	}
458
459	uint64_t taosHton64(uint64_t val) {	358,295,724✔
460	#if defined(WINDOWS) \|\| defined(DARWIN)
461	return ((val & 0x00000000000000ff) << 7 * 8) \| ((val & 0x000000000000ff00) << 5 * 8) \|
462	((val & 0x0000000000ff0000) << 3 * 8) \| ((val & 0x00000000ff000000) << 1 * 8) \|
463	((val & 0x000000ff00000000) >> 1 * 8) \| ((val & 0x0000ff0000000000) >> 3 * 8) \|
464	((val & 0x00ff000000000000) >> 5 * 8) \| ((val & 0xff00000000000000) >> 7 * 8);
465	#else
466	if (__BYTE_ORDER == __LITTLE_ENDIAN) {
467	return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) \| (unsigned int)htonl((int)(val >> 32));	358,295,724✔
468	} else if (__BYTE_ORDER == __BIG_ENDIAN) {
469	return val;
470	}
471	#endif
472	}
473
474	uint64_t taosNtoh64(uint64_t val) {	1,433,339✔
475	#if defined(WINDOWS) \|\| defined(DARWIN)
476	return taosHton64(val);
477	#else
478	if (__BYTE_ORDER == __LITTLE_ENDIAN) {
479	return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) \| (unsigned int)htonl((int)(val >> 32));	1,433,339✔
480	} else if (__BYTE_ORDER == __BIG_ENDIAN) {
481	return val;
482	}
483	#endif
484	}
485
486	int32_t taosSetSockOpt2(int32_t fd) {	69,580,988✔
487	#if defined(WINDOWS) \|\| defined(DARWIN)
488	return 0;
489	#else
490	int32_t ret = setsockopt(fd, IPPROTO_TCP, TCP_QUICKACK, (int[]){1}, sizeof(int));	69,580,988✔
491	if (ret < 0) {	69,670,908!
492	terrno = TAOS_SYSTEM_ERROR(errno);	×
493	return terrno;	×
494	} else {
495	return 0;	69,687,580✔
496	}
497	#endif
498	return 0;
499	}

taosdata / TDengine / #3522

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous