1 /*
2 *
3 * Copyright 2018 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 #include <assert.h>
24 #include <string.h>
25
26 #include <string>
27
28 #include <grpc/support/alloc.h>
29 #include <grpc/support/log.h>
30
31 #include "src/core/lib/channel/channel_args.h"
32 #include "src/core/lib/iomgr/error.h"
33 #include "src/core/lib/iomgr/exec_ctx.h"
34 #include "src/core/lib/iomgr/iomgr_custom.h"
35 #include "src/core/lib/iomgr/sockaddr.h"
36 #include "src/core/lib/iomgr/sockaddr_utils.h"
37 #include "src/core/lib/iomgr/tcp_custom.h"
38 #include "src/core/lib/iomgr/tcp_server.h"
39
40 extern grpc_core::TraceFlag grpc_tcp_trace;
41
42 extern grpc_socket_vtable* grpc_custom_socket_vtable;
43
44 /* one listening port */
45 struct grpc_tcp_listener {
46 grpc_tcp_server* server;
47 unsigned port_index;
48 int port;
49
50 grpc_custom_socket* socket;
51
52 /* linked list */
53 struct grpc_tcp_listener* next;
54
55 bool closed;
56 };
57
58 struct grpc_tcp_server {
59 gpr_refcount refs;
60
61 /* Called whenever accept() succeeds on a server port. */
62 grpc_tcp_server_cb on_accept_cb;
63 void* on_accept_cb_arg;
64
65 int open_ports;
66
67 /* linked list of server ports */
68 grpc_tcp_listener* head;
69 grpc_tcp_listener* tail;
70
71 /* List of closures passed to shutdown_starting_add(). */
72 grpc_closure_list shutdown_starting;
73
74 /* shutdown callback */
75 grpc_closure* shutdown_complete;
76
77 bool shutdown;
78 bool so_reuseport;
79
80 grpc_resource_quota* resource_quota;
81 };
82
tcp_server_create(grpc_closure * shutdown_complete,const grpc_channel_args * args,grpc_tcp_server ** server)83 static grpc_error* tcp_server_create(grpc_closure* shutdown_complete,
84 const grpc_channel_args* args,
85 grpc_tcp_server** server) {
86 grpc_tcp_server* s =
87 static_cast<grpc_tcp_server*>(gpr_malloc(sizeof(grpc_tcp_server)));
88 // Let the implementation decide if so_reuseport can be enabled or not.
89 s->so_reuseport = true;
90 s->resource_quota = grpc_resource_quota_create(nullptr);
91 for (size_t i = 0; i < (args == nullptr ? 0 : args->num_args); i++) {
92 if (!grpc_channel_args_find_bool(args, GRPC_ARG_ALLOW_REUSEPORT, true)) {
93 s->so_reuseport = false;
94 }
95 if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) {
96 if (args->args[i].type == GRPC_ARG_POINTER) {
97 grpc_resource_quota_unref_internal(s->resource_quota);
98 s->resource_quota = grpc_resource_quota_ref_internal(
99 static_cast<grpc_resource_quota*>(args->args[i].value.pointer.p));
100 } else {
101 grpc_resource_quota_unref_internal(s->resource_quota);
102 gpr_free(s);
103 return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
104 GRPC_ARG_RESOURCE_QUOTA " must be a pointer to a buffer pool");
105 }
106 }
107 }
108 gpr_ref_init(&s->refs, 1);
109 s->on_accept_cb = nullptr;
110 s->on_accept_cb_arg = nullptr;
111 s->open_ports = 0;
112 s->head = nullptr;
113 s->tail = nullptr;
114 s->shutdown_starting.head = nullptr;
115 s->shutdown_starting.tail = nullptr;
116 s->shutdown_complete = shutdown_complete;
117 s->shutdown = false;
118 *server = s;
119 return GRPC_ERROR_NONE;
120 }
121
tcp_server_ref(grpc_tcp_server * s)122 static grpc_tcp_server* tcp_server_ref(grpc_tcp_server* s) {
123 GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
124 gpr_ref(&s->refs);
125 return s;
126 }
127
tcp_server_shutdown_starting_add(grpc_tcp_server * s,grpc_closure * shutdown_starting)128 static void tcp_server_shutdown_starting_add(grpc_tcp_server* s,
129 grpc_closure* shutdown_starting) {
130 grpc_closure_list_append(&s->shutdown_starting, shutdown_starting,
131 GRPC_ERROR_NONE);
132 }
133
finish_shutdown(grpc_tcp_server * s)134 static void finish_shutdown(grpc_tcp_server* s) {
135 GPR_ASSERT(s->shutdown);
136 if (s->shutdown_complete != nullptr) {
137 grpc_core::ExecCtx::Run(DEBUG_LOCATION, s->shutdown_complete,
138 GRPC_ERROR_NONE);
139 }
140
141 while (s->head) {
142 grpc_tcp_listener* sp = s->head;
143 s->head = sp->next;
144 sp->next = nullptr;
145 gpr_free(sp);
146 }
147 grpc_resource_quota_unref_internal(s->resource_quota);
148 gpr_free(s);
149 }
150
custom_close_callback(grpc_custom_socket * socket)151 static void custom_close_callback(grpc_custom_socket* socket) {
152 grpc_tcp_listener* sp = socket->listener;
153 if (sp) {
154 grpc_core::ExecCtx exec_ctx;
155 sp->server->open_ports--;
156 if (sp->server->open_ports == 0 && sp->server->shutdown) {
157 finish_shutdown(sp->server);
158 }
159 }
160 socket->refs--;
161 if (socket->refs == 0) {
162 grpc_custom_socket_vtable->destroy(socket);
163 gpr_free(socket);
164 }
165 }
166
grpc_custom_close_server_callback(grpc_tcp_listener * listener)167 void grpc_custom_close_server_callback(grpc_tcp_listener* listener) {
168 if (listener) {
169 grpc_core::ExecCtx exec_ctx;
170 listener->server->open_ports--;
171 if (listener->server->open_ports == 0 && listener->server->shutdown) {
172 finish_shutdown(listener->server);
173 }
174 }
175 }
176
close_listener(grpc_tcp_listener * sp)177 static void close_listener(grpc_tcp_listener* sp) {
178 grpc_custom_socket* socket = sp->socket;
179 if (!sp->closed) {
180 sp->closed = true;
181 grpc_custom_socket_vtable->close(socket, custom_close_callback);
182 }
183 }
184
tcp_server_destroy(grpc_tcp_server * s)185 static void tcp_server_destroy(grpc_tcp_server* s) {
186 int immediately_done = 0;
187 grpc_tcp_listener* sp;
188
189 GPR_ASSERT(!s->shutdown);
190 s->shutdown = true;
191
192 if (s->open_ports == 0) {
193 immediately_done = 1;
194 }
195 for (sp = s->head; sp; sp = sp->next) {
196 close_listener(sp);
197 }
198
199 if (immediately_done) {
200 finish_shutdown(s);
201 }
202 }
203
tcp_server_unref(grpc_tcp_server * s)204 static void tcp_server_unref(grpc_tcp_server* s) {
205 GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
206 if (gpr_unref(&s->refs)) {
207 /* Complete shutdown_starting work before destroying. */
208 grpc_core::ExecCtx exec_ctx;
209 grpc_core::ExecCtx::RunList(DEBUG_LOCATION, &s->shutdown_starting);
210 grpc_core::ExecCtx::Get()->Flush();
211 tcp_server_destroy(s);
212 }
213 }
214
finish_accept(grpc_tcp_listener * sp,grpc_custom_socket * socket)215 static void finish_accept(grpc_tcp_listener* sp, grpc_custom_socket* socket) {
216 grpc_tcp_server_acceptor* acceptor =
217 static_cast<grpc_tcp_server_acceptor*>(gpr_malloc(sizeof(*acceptor)));
218 grpc_endpoint* ep = nullptr;
219 grpc_resolved_address peer_name;
220 std::string peer_name_string;
221 grpc_error* err;
222
223 memset(&peer_name, 0, sizeof(grpc_resolved_address));
224 peer_name.len = GRPC_MAX_SOCKADDR_SIZE;
225 err = grpc_custom_socket_vtable->getpeername(
226 socket, reinterpret_cast<grpc_sockaddr*>(&peer_name.addr),
227 reinterpret_cast<int*>(&peer_name.len));
228 if (err == GRPC_ERROR_NONE) {
229 peer_name_string = grpc_sockaddr_to_uri(&peer_name);
230 } else {
231 GRPC_LOG_IF_ERROR("getpeername error", err);
232 GRPC_ERROR_UNREF(err);
233 }
234 if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
235 gpr_log(GPR_INFO, "SERVER_CONNECT: %p accepted connection: %s", sp->server,
236 peer_name_string.c_str());
237 }
238 ep = custom_tcp_endpoint_create(socket, sp->server->resource_quota,
239 peer_name_string.c_str());
240 acceptor->from_server = sp->server;
241 acceptor->port_index = sp->port_index;
242 acceptor->fd_index = 0;
243 acceptor->external_connection = false;
244 sp->server->on_accept_cb(sp->server->on_accept_cb_arg, ep, nullptr, acceptor);
245 }
246
247 static void custom_accept_callback(grpc_custom_socket* socket,
248 grpc_custom_socket* client,
249 grpc_error* error);
250
custom_accept_callback(grpc_custom_socket * socket,grpc_custom_socket * client,grpc_error * error)251 static void custom_accept_callback(grpc_custom_socket* socket,
252 grpc_custom_socket* client,
253 grpc_error* error) {
254 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
255 grpc_core::ExecCtx exec_ctx;
256 grpc_tcp_listener* sp = socket->listener;
257 if (error != GRPC_ERROR_NONE) {
258 if (!sp->closed) {
259 gpr_log(GPR_ERROR, "Accept failed: %s", grpc_error_string(error));
260 }
261 gpr_free(client);
262 GRPC_ERROR_UNREF(error);
263 return;
264 }
265 finish_accept(sp, client);
266 if (!sp->closed) {
267 grpc_custom_socket* new_socket = static_cast<grpc_custom_socket*>(
268 gpr_malloc(sizeof(grpc_custom_socket)));
269 new_socket->endpoint = nullptr;
270 new_socket->listener = nullptr;
271 new_socket->connector = nullptr;
272 new_socket->refs = 1;
273 grpc_custom_socket_vtable->accept(sp->socket, new_socket,
274 custom_accept_callback);
275 }
276 }
277
add_socket_to_server(grpc_tcp_server * s,grpc_custom_socket * socket,const grpc_resolved_address * addr,unsigned port_index,grpc_tcp_listener ** listener)278 static grpc_error* add_socket_to_server(grpc_tcp_server* s,
279 grpc_custom_socket* socket,
280 const grpc_resolved_address* addr,
281 unsigned port_index,
282 grpc_tcp_listener** listener) {
283 grpc_tcp_listener* sp = nullptr;
284 int port = -1;
285 grpc_error* error;
286 grpc_resolved_address sockname_temp;
287
288 // NOTE(lidiz) The last argument is "flags" which is unused by other
289 // implementations. Python IO managers uses it to specify SO_REUSEPORT.
290 int flags = 0;
291 if (s->so_reuseport) {
292 flags |= GRPC_CUSTOM_SOCKET_OPT_SO_REUSEPORT;
293 }
294
295 error = grpc_custom_socket_vtable->bind(
296 socket, reinterpret_cast<grpc_sockaddr*>(const_cast<char*>(addr->addr)),
297 addr->len, flags);
298 if (error != GRPC_ERROR_NONE) {
299 return error;
300 }
301
302 error = grpc_custom_socket_vtable->listen(socket);
303 if (error != GRPC_ERROR_NONE) {
304 return error;
305 }
306
307 sockname_temp.len = GRPC_MAX_SOCKADDR_SIZE;
308 error = grpc_custom_socket_vtable->getsockname(
309 socket, reinterpret_cast<grpc_sockaddr*>(&sockname_temp.addr),
310 reinterpret_cast<int*>(&sockname_temp.len));
311 if (error != GRPC_ERROR_NONE) {
312 return error;
313 }
314
315 port = grpc_sockaddr_get_port(&sockname_temp);
316
317 GPR_ASSERT(port >= 0);
318 GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");
319 sp = static_cast<grpc_tcp_listener*>(gpr_zalloc(sizeof(grpc_tcp_listener)));
320 sp->next = nullptr;
321 if (s->head == nullptr) {
322 s->head = sp;
323 } else {
324 s->tail->next = sp;
325 }
326 s->tail = sp;
327 sp->server = s;
328 sp->socket = socket;
329 sp->port = port;
330 sp->port_index = port_index;
331 sp->closed = false;
332 s->open_ports++;
333 *listener = sp;
334
335 return GRPC_ERROR_NONE;
336 }
337
tcp_server_add_port(grpc_tcp_server * s,const grpc_resolved_address * addr,int * port)338 static grpc_error* tcp_server_add_port(grpc_tcp_server* s,
339 const grpc_resolved_address* addr,
340 int* port) {
341 // This function is mostly copied from tcp_server_windows.c
342 grpc_tcp_listener* sp = nullptr;
343 grpc_custom_socket* socket;
344 grpc_resolved_address addr6_v4mapped;
345 grpc_resolved_address wildcard;
346 grpc_resolved_address* allocated_addr = nullptr;
347 grpc_resolved_address sockname_temp;
348 unsigned port_index = 0;
349 grpc_error* error = GRPC_ERROR_NONE;
350 int family;
351
352 GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
353
354 if (s->tail != nullptr) {
355 port_index = s->tail->port_index + 1;
356 }
357
358 /* Check if this is a wildcard port, and if so, try to keep the port the same
359 as some previously created listener. */
360 if (grpc_sockaddr_get_port(addr) == 0) {
361 for (sp = s->head; sp; sp = sp->next) {
362 socket = sp->socket;
363 sockname_temp.len = GRPC_MAX_SOCKADDR_SIZE;
364 if (nullptr == grpc_custom_socket_vtable->getsockname(
365 socket,
366 reinterpret_cast<grpc_sockaddr*>(&sockname_temp.addr),
367 reinterpret_cast<int*>(&sockname_temp.len))) {
368 *port = grpc_sockaddr_get_port(&sockname_temp);
369 if (*port > 0) {
370 allocated_addr = static_cast<grpc_resolved_address*>(
371 gpr_malloc(sizeof(grpc_resolved_address)));
372 memcpy(allocated_addr, addr, sizeof(grpc_resolved_address));
373 grpc_sockaddr_set_port(allocated_addr, *port);
374 addr = allocated_addr;
375 break;
376 }
377 }
378 }
379 }
380
381 if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
382 addr = &addr6_v4mapped;
383 }
384
385 /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
386 if (grpc_sockaddr_is_wildcard(addr, port)) {
387 grpc_sockaddr_make_wildcard6(*port, &wildcard);
388
389 addr = &wildcard;
390 }
391
392 if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
393 gpr_log(GPR_INFO, "SERVER %p add_port %s error=%s", s,
394 grpc_sockaddr_to_string(addr, false).c_str(),
395 grpc_error_string(error));
396 }
397
398 family = grpc_sockaddr_get_family(addr);
399 socket =
400 static_cast<grpc_custom_socket*>(gpr_malloc(sizeof(grpc_custom_socket)));
401 socket->refs = 1;
402 socket->endpoint = nullptr;
403 socket->listener = nullptr;
404 socket->connector = nullptr;
405 error = grpc_custom_socket_vtable->init(socket, family);
406
407 if (error == GRPC_ERROR_NONE) {
408 error = add_socket_to_server(s, socket, addr, port_index, &sp);
409 }
410 gpr_free(allocated_addr);
411
412 if (error != GRPC_ERROR_NONE) {
413 grpc_error* error_out = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
414 "Failed to add port to server", &error, 1);
415 GRPC_ERROR_UNREF(error);
416 error = error_out;
417 *port = -1;
418 } else {
419 GPR_ASSERT(sp != nullptr);
420 *port = sp->port;
421 }
422 socket->listener = sp;
423 return error;
424 }
425
tcp_server_start(grpc_tcp_server * server,const std::vector<grpc_pollset * > *,grpc_tcp_server_cb on_accept_cb,void * cb_arg)426 static void tcp_server_start(grpc_tcp_server* server,
427 const std::vector<grpc_pollset*>* /*pollsets*/,
428 grpc_tcp_server_cb on_accept_cb, void* cb_arg) {
429 grpc_tcp_listener* sp;
430 GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
431 if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
432 gpr_log(GPR_INFO, "SERVER_START %p", server);
433 }
434 GPR_ASSERT(on_accept_cb);
435 GPR_ASSERT(!server->on_accept_cb);
436 server->on_accept_cb = on_accept_cb;
437 server->on_accept_cb_arg = cb_arg;
438 for (sp = server->head; sp; sp = sp->next) {
439 grpc_custom_socket* new_socket = static_cast<grpc_custom_socket*>(
440 gpr_malloc(sizeof(grpc_custom_socket)));
441 new_socket->endpoint = nullptr;
442 new_socket->listener = nullptr;
443 new_socket->connector = nullptr;
444 new_socket->refs = 1;
445 grpc_custom_socket_vtable->accept(sp->socket, new_socket,
446 custom_accept_callback);
447 }
448 }
449
tcp_server_port_fd_count(grpc_tcp_server *,unsigned)450 static unsigned tcp_server_port_fd_count(grpc_tcp_server* /*s*/,
451 unsigned /*port_index*/) {
452 return 0;
453 }
454
tcp_server_port_fd(grpc_tcp_server *,unsigned,unsigned)455 static int tcp_server_port_fd(grpc_tcp_server* /*s*/, unsigned /*port_index*/,
456 unsigned /*fd_index*/) {
457 return -1;
458 }
459
tcp_server_shutdown_listeners(grpc_tcp_server * s)460 static void tcp_server_shutdown_listeners(grpc_tcp_server* s) {
461 for (grpc_tcp_listener* sp = s->head; sp; sp = sp->next) {
462 if (!sp->closed) {
463 sp->closed = true;
464 grpc_custom_socket_vtable->close(sp->socket, custom_close_callback);
465 }
466 }
467 }
468
tcp_server_create_fd_handler(grpc_tcp_server *)469 static grpc_core::TcpServerFdHandler* tcp_server_create_fd_handler(
470 grpc_tcp_server* /*s*/) {
471 return nullptr;
472 }
473
474 grpc_tcp_server_vtable custom_tcp_server_vtable = {
475 tcp_server_create, tcp_server_start,
476 tcp_server_add_port, tcp_server_create_fd_handler,
477 tcp_server_port_fd_count, tcp_server_port_fd,
478 tcp_server_ref, tcp_server_shutdown_starting_add,
479 tcp_server_unref, tcp_server_shutdown_listeners};
480
481 #ifdef GRPC_UV_TEST
482 grpc_tcp_server_vtable* default_tcp_server_vtable = &custom_tcp_server_vtable;
483 #endif
484