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