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