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