1 /*
2 *
3 * Copyright 2017 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include <grpc/support/port_platform.h>
20
21 #include "src/core/lib/iomgr/port.h"
22
23 #ifdef GRPC_POSIX_SOCKET_TCP_SERVER_UTILS_COMMON
24
25 #include "src/core/lib/iomgr/tcp_server_utils_posix.h"
26
27 #include <errno.h>
28 #include <limits.h>
29 #include <stdio.h>
30 #include <string.h>
31
32 #include <string>
33
34 #include "absl/strings/str_cat.h"
35
36 #include <grpc/support/alloc.h>
37 #include <grpc/support/log.h>
38 #include <grpc/support/sync.h>
39
40 #include "src/core/lib/iomgr/error.h"
41 #include "src/core/lib/iomgr/sockaddr.h"
42 #include "src/core/lib/iomgr/sockaddr_utils.h"
43 #include "src/core/lib/iomgr/unix_sockets_posix.h"
44
45 #define MIN_SAFE_ACCEPT_QUEUE_SIZE 100
46
47 static gpr_once s_init_max_accept_queue_size = GPR_ONCE_INIT;
48 static int s_max_accept_queue_size;
49
50 /* get max listen queue size on linux */
init_max_accept_queue_size(void)51 static void init_max_accept_queue_size(void) {
52 int n = SOMAXCONN;
53 char buf[64];
54 FILE* fp = fopen("/proc/sys/net/core/somaxconn", "r");
55 if (fp == nullptr) {
56 /* 2.4 kernel. */
57 s_max_accept_queue_size = SOMAXCONN;
58 return;
59 }
60 if (fgets(buf, sizeof buf, fp)) {
61 char* end;
62 long i = strtol(buf, &end, 10);
63 if (i > 0 && i <= INT_MAX && end && *end == '\n') {
64 n = static_cast<int>(i);
65 }
66 }
67 fclose(fp);
68 s_max_accept_queue_size = n;
69
70 if (s_max_accept_queue_size < MIN_SAFE_ACCEPT_QUEUE_SIZE) {
71 gpr_log(GPR_INFO,
72 "Suspiciously small accept queue (%d) will probably lead to "
73 "connection drops",
74 s_max_accept_queue_size);
75 }
76 }
77
get_max_accept_queue_size(void)78 static int get_max_accept_queue_size(void) {
79 gpr_once_init(&s_init_max_accept_queue_size, init_max_accept_queue_size);
80 return s_max_accept_queue_size;
81 }
82
add_socket_to_server(grpc_tcp_server * s,int fd,const grpc_resolved_address * addr,unsigned port_index,unsigned fd_index,grpc_tcp_listener ** listener)83 static grpc_error* add_socket_to_server(grpc_tcp_server* s, int fd,
84 const grpc_resolved_address* addr,
85 unsigned port_index, unsigned fd_index,
86 grpc_tcp_listener** listener) {
87 grpc_tcp_listener* sp = nullptr;
88 int port = -1;
89
90 grpc_error* err =
91 grpc_tcp_server_prepare_socket(s, fd, addr, s->so_reuseport, &port);
92 if (err == GRPC_ERROR_NONE) {
93 GPR_ASSERT(port > 0);
94 std::string addr_str = grpc_sockaddr_to_string(addr, true);
95 std::string name = absl::StrCat("tcp-server-listener:", addr_str);
96 gpr_mu_lock(&s->mu);
97 s->nports++;
98 GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");
99 sp = static_cast<grpc_tcp_listener*>(gpr_malloc(sizeof(grpc_tcp_listener)));
100 sp->next = nullptr;
101 if (s->head == nullptr) {
102 s->head = sp;
103 } else {
104 s->tail->next = sp;
105 }
106 s->tail = sp;
107 sp->server = s;
108 sp->fd = fd;
109 sp->emfd = grpc_fd_create(fd, name.c_str(), true);
110 memcpy(&sp->addr, addr, sizeof(grpc_resolved_address));
111 sp->port = port;
112 sp->port_index = port_index;
113 sp->fd_index = fd_index;
114 sp->is_sibling = 0;
115 sp->sibling = nullptr;
116 GPR_ASSERT(sp->emfd);
117 gpr_mu_unlock(&s->mu);
118 }
119
120 *listener = sp;
121 return err;
122 }
123
124 /* If successful, add a listener to s for addr, set *dsmode for the socket, and
125 return the *listener. */
grpc_tcp_server_add_addr(grpc_tcp_server * s,const grpc_resolved_address * addr,unsigned port_index,unsigned fd_index,grpc_dualstack_mode * dsmode,grpc_tcp_listener ** listener)126 grpc_error* grpc_tcp_server_add_addr(grpc_tcp_server* s,
127 const grpc_resolved_address* addr,
128 unsigned port_index, unsigned fd_index,
129 grpc_dualstack_mode* dsmode,
130 grpc_tcp_listener** listener) {
131 grpc_resolved_address addr4_copy;
132 int fd;
133 grpc_error* err =
134 grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd);
135 if (err != GRPC_ERROR_NONE) {
136 return err;
137 }
138 if (*dsmode == GRPC_DSMODE_IPV4 &&
139 grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
140 addr = &addr4_copy;
141 }
142 return add_socket_to_server(s, fd, addr, port_index, fd_index, listener);
143 }
144
145 /* Prepare a recently-created socket for listening. */
grpc_tcp_server_prepare_socket(grpc_tcp_server * s,int fd,const grpc_resolved_address * addr,bool so_reuseport,int * port)146 grpc_error* grpc_tcp_server_prepare_socket(grpc_tcp_server* s, int fd,
147 const grpc_resolved_address* addr,
148 bool so_reuseport, int* port) {
149 grpc_resolved_address sockname_temp;
150 grpc_error* err = GRPC_ERROR_NONE;
151
152 GPR_ASSERT(fd >= 0);
153
154 if (so_reuseport && !grpc_is_unix_socket(addr)) {
155 err = grpc_set_socket_reuse_port(fd, 1);
156 if (err != GRPC_ERROR_NONE) goto error;
157 }
158
159 #ifdef GRPC_LINUX_ERRQUEUE
160 err = grpc_set_socket_zerocopy(fd);
161 if (err != GRPC_ERROR_NONE) {
162 /* it's not fatal, so just log it. */
163 gpr_log(GPR_DEBUG, "Node does not support SO_ZEROCOPY, continuing.");
164 GRPC_ERROR_UNREF(err);
165 }
166 #endif
167 err = grpc_set_socket_nonblocking(fd, 1);
168 if (err != GRPC_ERROR_NONE) goto error;
169 err = grpc_set_socket_cloexec(fd, 1);
170 if (err != GRPC_ERROR_NONE) goto error;
171 if (!grpc_is_unix_socket(addr)) {
172 err = grpc_set_socket_low_latency(fd, 1);
173 if (err != GRPC_ERROR_NONE) goto error;
174 err = grpc_set_socket_reuse_addr(fd, 1);
175 if (err != GRPC_ERROR_NONE) goto error;
176 err = grpc_set_socket_tcp_user_timeout(fd, s->channel_args,
177 false /* is_client */);
178 if (err != GRPC_ERROR_NONE) goto error;
179 }
180 err = grpc_set_socket_no_sigpipe_if_possible(fd);
181 if (err != GRPC_ERROR_NONE) goto error;
182
183 err = grpc_apply_socket_mutator_in_args(fd, s->channel_args);
184 if (err != GRPC_ERROR_NONE) goto error;
185
186 if (bind(fd, reinterpret_cast<grpc_sockaddr*>(const_cast<char*>(addr->addr)),
187 addr->len) < 0) {
188 err = GRPC_OS_ERROR(errno, "bind");
189 goto error;
190 }
191
192 if (listen(fd, get_max_accept_queue_size()) < 0) {
193 err = GRPC_OS_ERROR(errno, "listen");
194 goto error;
195 }
196
197 sockname_temp.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
198
199 if (getsockname(fd, reinterpret_cast<grpc_sockaddr*>(sockname_temp.addr),
200 &sockname_temp.len) < 0) {
201 err = GRPC_OS_ERROR(errno, "getsockname");
202 goto error;
203 }
204
205 *port = grpc_sockaddr_get_port(&sockname_temp);
206 return GRPC_ERROR_NONE;
207
208 error:
209 GPR_ASSERT(err != GRPC_ERROR_NONE);
210 if (fd >= 0) {
211 close(fd);
212 }
213 grpc_error* ret =
214 grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
215 "Unable to configure socket", &err, 1),
216 GRPC_ERROR_INT_FD, fd);
217 GRPC_ERROR_UNREF(err);
218 return ret;
219 }
220
221 #endif /* GRPC_POSIX_SOCKET_TCP_SERVER_UTILS_COMMON */
222