1 /*
2 *
3 * Copyright 2015 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 /* FIXME: "posix" files shouldn't be depending on _GNU_SOURCE */
20 #ifndef _GNU_SOURCE
21 #define _GNU_SOURCE
22 #endif
23
24 #ifndef SO_RXQ_OVFL
25 #define SO_RXQ_OVFL 40
26 #endif
27
28 #include <grpc/support/port_platform.h>
29
30 #include "src/core/lib/iomgr/port.h"
31
32 #ifdef GRPC_POSIX_SOCKET
33
34 #include "src/core/lib/iomgr/udp_server.h"
35
36 #include <errno.h>
37 #include <fcntl.h>
38 #include <limits.h>
39 #include <netinet/in.h>
40 #include <netinet/tcp.h>
41 #include <string.h>
42 #include <sys/socket.h>
43 #include <sys/stat.h>
44 #include <sys/types.h>
45 #include <unistd.h>
46
47 #include <grpc/grpc.h>
48 #include <grpc/support/alloc.h>
49 #include <grpc/support/log.h>
50 #include <grpc/support/string_util.h>
51 #include <grpc/support/sync.h>
52 #include <grpc/support/time.h>
53 #include "src/core/lib/channel/channel_args.h"
54 #include "src/core/lib/gpr/string.h"
55 #include "src/core/lib/gprpp/inlined_vector.h"
56 #include "src/core/lib/gprpp/memory.h"
57 #include "src/core/lib/iomgr/error.h"
58 #include "src/core/lib/iomgr/ev_posix.h"
59 #include "src/core/lib/iomgr/executor.h"
60 #include "src/core/lib/iomgr/resolve_address.h"
61 #include "src/core/lib/iomgr/sockaddr.h"
62 #include "src/core/lib/iomgr/sockaddr_utils.h"
63 #include "src/core/lib/iomgr/socket_factory_posix.h"
64 #include "src/core/lib/iomgr/socket_utils_posix.h"
65 #include "src/core/lib/iomgr/unix_sockets_posix.h"
66
67 /* A listener which implements basic features of Listening on a port for
68 * I/O events*/
69 class GrpcUdpListener {
70 public:
71 GrpcUdpListener(grpc_udp_server* server, int fd,
72 const grpc_resolved_address* addr);
73 ~GrpcUdpListener();
74
75 /* Called when grpc server starts to listening on the grpc_fd. */
76 void StartListening(grpc_pollset** pollsets, size_t pollset_count,
77 GrpcUdpHandlerFactory* handler_factory);
78
79 /* Called when data is available to read from the socket.
80 * Return true if there is more data to read from fd. */
81 void OnRead(grpc_error* error, void* do_read_arg);
82
83 /* Called when the socket is writeable. The given closure should be scheduled
84 * when the socket becomes blocked next time. */
85 void OnCanWrite(grpc_error* error, void* do_write_arg);
86
87 /* Called when the grpc_fd is about to be orphaned (and the FD closed). */
88 void OnFdAboutToOrphan();
89
90 /* Called to orphan fd of this listener.*/
91 void OrphanFd();
92
93 /* Called when this listener is going to be destroyed. */
94 void OnDestroy();
95
fd() const96 int fd() const { return fd_; }
97
98 protected:
emfd() const99 grpc_fd* emfd() const { return emfd_; }
100
mutex()101 gpr_mu* mutex() { return &mutex_; }
102
103 private:
104 /* event manager callback when reads are ready */
105 static void on_read(void* arg, grpc_error* error);
106 static void on_write(void* arg, grpc_error* error);
107
108 static void do_read(void* arg, grpc_error* error);
109 static void do_write(void* arg, grpc_error* error);
110 // Wrapper of grpc_fd_notify_on_write() with a grpc_closure callback
111 // interface.
112 static void fd_notify_on_write_wrapper(void* arg, grpc_error* error);
113
114 static void shutdown_fd(void* args, grpc_error* error);
115
116 int fd_;
117 grpc_fd* emfd_;
118 grpc_udp_server* server_;
119 grpc_resolved_address addr_;
120 grpc_closure read_closure_;
121 grpc_closure write_closure_;
122 // To be called when corresponding QuicGrpcServer closes all active
123 // connections.
124 grpc_closure orphan_fd_closure_;
125 grpc_closure destroyed_closure_;
126 // To be scheduled on another thread to actually read/write.
127 grpc_closure do_read_closure_;
128 grpc_closure do_write_closure_;
129 grpc_closure notify_on_write_closure_;
130 // True if orphan_cb is trigered.
131 bool orphan_notified_;
132 // True if grpc_fd_notify_on_write() is called after on_write() call.
133 bool notify_on_write_armed_;
134 // True if fd has been shutdown.
135 bool already_shutdown_;
136 // Object actually handles I/O events. Assigned in StartListening().
137 GrpcUdpHandler* udp_handler_ = nullptr;
138 // To be notified on destruction.
139 GrpcUdpHandlerFactory* handler_factory_ = nullptr;
140 // Required to access above fields.
141 gpr_mu mutex_;
142 };
143
GrpcUdpListener(grpc_udp_server * server,int fd,const grpc_resolved_address * addr)144 GrpcUdpListener::GrpcUdpListener(grpc_udp_server* server, int fd,
145 const grpc_resolved_address* addr)
146 : fd_(fd),
147 server_(server),
148 orphan_notified_(false),
149 already_shutdown_(false) {
150 char* addr_str;
151 char* name;
152 grpc_sockaddr_to_string(&addr_str, addr, 1);
153 gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
154 gpr_free(addr_str);
155 emfd_ = grpc_fd_create(fd, name, true);
156 memcpy(&addr_, addr, sizeof(grpc_resolved_address));
157 GPR_ASSERT(emfd_);
158 gpr_free(name);
159 gpr_mu_init(&mutex_);
160 }
161
~GrpcUdpListener()162 GrpcUdpListener::~GrpcUdpListener() { gpr_mu_destroy(&mutex_); }
163
164 /* the overall server */
165 struct grpc_udp_server {
166 gpr_mu mu;
167
168 /* factory to use for creating and binding sockets, or NULL */
169 grpc_socket_factory* socket_factory;
170
171 /* active port count: how many ports are actually still listening */
172 size_t active_ports;
173 /* destroyed port count: how many ports are completely destroyed */
174 size_t destroyed_ports;
175
176 /* is this server shutting down? (boolean) */
177 int shutdown;
178
179 /* An array of listeners */
180 grpc_core::InlinedVector<GrpcUdpListener, 16> listeners;
181
182 /* factory for use to create udp listeners */
183 GrpcUdpHandlerFactory* handler_factory;
184
185 /* shutdown callback */
186 grpc_closure* shutdown_complete;
187
188 /* all pollsets interested in new connections */
189 grpc_pollset** pollsets;
190 /* number of pollsets in the pollsets array */
191 size_t pollset_count;
192 /* opaque object to pass to callbacks */
193 void* user_data;
194
195 /* latch has_so_reuseport during server creation */
196 bool so_reuseport;
197 };
198
get_socket_factory(const grpc_channel_args * args)199 static grpc_socket_factory* get_socket_factory(const grpc_channel_args* args) {
200 if (args) {
201 const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_SOCKET_FACTORY);
202 if (arg) {
203 GPR_ASSERT(arg->type == GRPC_ARG_POINTER);
204 return static_cast<grpc_socket_factory*>(arg->value.pointer.p);
205 }
206 }
207 return nullptr;
208 }
209
grpc_udp_server_create(const grpc_channel_args * args)210 grpc_udp_server* grpc_udp_server_create(const grpc_channel_args* args) {
211 grpc_udp_server* s = grpc_core::New<grpc_udp_server>();
212 gpr_mu_init(&s->mu);
213 s->socket_factory = get_socket_factory(args);
214 if (s->socket_factory) {
215 grpc_socket_factory_ref(s->socket_factory);
216 }
217 s->active_ports = 0;
218 s->destroyed_ports = 0;
219 s->shutdown = 0;
220 s->so_reuseport = grpc_is_socket_reuse_port_supported();
221 return s;
222 }
223
224 // static
shutdown_fd(void * args,grpc_error * error)225 void GrpcUdpListener::shutdown_fd(void* args, grpc_error* error) {
226 if (args == nullptr) {
227 // No-op if shutdown args are null.
228 return;
229 }
230 auto sp = static_cast<GrpcUdpListener*>(args);
231 gpr_mu_lock(sp->mutex());
232 gpr_log(GPR_DEBUG, "shutdown fd %d", sp->fd_);
233 grpc_fd_shutdown(sp->emfd_, GRPC_ERROR_REF(error));
234 sp->already_shutdown_ = true;
235 if (!sp->notify_on_write_armed_) {
236 // Re-arm write notification to notify listener with error. This is
237 // necessary to decrement active_ports.
238 sp->notify_on_write_armed_ = true;
239 grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
240 }
241 gpr_mu_unlock(sp->mutex());
242 }
243
finish_shutdown(grpc_udp_server * s)244 static void finish_shutdown(grpc_udp_server* s) {
245 if (s->shutdown_complete != nullptr) {
246 GRPC_CLOSURE_SCHED(s->shutdown_complete, GRPC_ERROR_NONE);
247 }
248
249 gpr_mu_destroy(&s->mu);
250
251 gpr_log(GPR_DEBUG, "Destroy all listeners.");
252 for (size_t i = 0; i < s->listeners.size(); ++i) {
253 s->listeners[i].OnDestroy();
254 }
255
256 if (s->socket_factory) {
257 grpc_socket_factory_unref(s->socket_factory);
258 }
259
260 grpc_core::Delete(s);
261 }
262
destroyed_port(void * server,grpc_error * error)263 static void destroyed_port(void* server, grpc_error* error) {
264 grpc_udp_server* s = static_cast<grpc_udp_server*>(server);
265 gpr_mu_lock(&s->mu);
266 s->destroyed_ports++;
267 if (s->destroyed_ports == s->listeners.size()) {
268 gpr_mu_unlock(&s->mu);
269 finish_shutdown(s);
270 } else {
271 gpr_mu_unlock(&s->mu);
272 }
273 }
274
275 /* called when all listening endpoints have been shutdown, so no further
276 events will be received on them - at this point it's safe to destroy
277 things */
deactivated_all_ports(grpc_udp_server * s)278 static void deactivated_all_ports(grpc_udp_server* s) {
279 /* delete ALL the things */
280 gpr_mu_lock(&s->mu);
281
282 GPR_ASSERT(s->shutdown);
283
284 if (s->listeners.size() == 0) {
285 gpr_mu_unlock(&s->mu);
286 finish_shutdown(s);
287 return;
288 }
289 for (size_t i = 0; i < s->listeners.size(); ++i) {
290 s->listeners[i].OrphanFd();
291 }
292 gpr_mu_unlock(&s->mu);
293 }
294
OrphanFd()295 void GrpcUdpListener::OrphanFd() {
296 gpr_log(GPR_DEBUG, "Orphan fd %d, emfd %p", fd_, emfd_);
297 grpc_unlink_if_unix_domain_socket(&addr_);
298
299 GRPC_CLOSURE_INIT(&destroyed_closure_, destroyed_port, server_,
300 grpc_schedule_on_exec_ctx);
301 /* Because at this point, all listening sockets have been shutdown already, no
302 * need to call OnFdAboutToOrphan() to notify the handler again. */
303 grpc_fd_orphan(emfd_, &destroyed_closure_, nullptr, "udp_listener_shutdown");
304 }
305
grpc_udp_server_destroy(grpc_udp_server * s,grpc_closure * on_done)306 void grpc_udp_server_destroy(grpc_udp_server* s, grpc_closure* on_done) {
307 gpr_mu_lock(&s->mu);
308
309 GPR_ASSERT(!s->shutdown);
310 s->shutdown = 1;
311
312 s->shutdown_complete = on_done;
313
314 gpr_log(GPR_DEBUG, "start to destroy udp_server");
315 /* shutdown all fd's */
316 if (s->active_ports) {
317 for (size_t i = 0; i < s->listeners.size(); ++i) {
318 GrpcUdpListener* sp = &s->listeners[i];
319 sp->OnFdAboutToOrphan();
320 }
321 gpr_mu_unlock(&s->mu);
322 } else {
323 gpr_mu_unlock(&s->mu);
324 deactivated_all_ports(s);
325 }
326 }
327
OnFdAboutToOrphan()328 void GrpcUdpListener::OnFdAboutToOrphan() {
329 gpr_mu_lock(&mutex_);
330 grpc_unlink_if_unix_domain_socket(&addr_);
331
332 GRPC_CLOSURE_INIT(&destroyed_closure_, destroyed_port, server_,
333 grpc_schedule_on_exec_ctx);
334 if (!orphan_notified_ && udp_handler_ != nullptr) {
335 /* Signals udp_handler that the FD is about to be closed and
336 * should no longer be used. */
337 GRPC_CLOSURE_INIT(&orphan_fd_closure_, shutdown_fd, this,
338 grpc_schedule_on_exec_ctx);
339 gpr_log(GPR_DEBUG, "fd %d about to be orphaned", fd_);
340 udp_handler_->OnFdAboutToOrphan(&orphan_fd_closure_, server_->user_data);
341 orphan_notified_ = true;
342 }
343 gpr_mu_unlock(&mutex_);
344 }
345
bind_socket(grpc_socket_factory * socket_factory,int sockfd,const grpc_resolved_address * addr)346 static int bind_socket(grpc_socket_factory* socket_factory, int sockfd,
347 const grpc_resolved_address* addr) {
348 return (socket_factory != nullptr)
349 ? grpc_socket_factory_bind(socket_factory, sockfd, addr)
350 : bind(sockfd,
351 reinterpret_cast<grpc_sockaddr*>(
352 const_cast<char*>(addr->addr)),
353 addr->len);
354 }
355
356 /* Prepare a recently-created socket for listening. */
prepare_socket(grpc_socket_factory * socket_factory,int fd,const grpc_resolved_address * addr,int rcv_buf_size,int snd_buf_size,bool so_reuseport)357 static int prepare_socket(grpc_socket_factory* socket_factory, int fd,
358 const grpc_resolved_address* addr, int rcv_buf_size,
359 int snd_buf_size, bool so_reuseport) {
360 grpc_resolved_address sockname_temp;
361 grpc_sockaddr* addr_ptr =
362 reinterpret_cast<grpc_sockaddr*>(const_cast<char*>(addr->addr));
363
364 if (fd < 0) {
365 goto error;
366 }
367
368 if (grpc_set_socket_nonblocking(fd, 1) != GRPC_ERROR_NONE) {
369 gpr_log(GPR_ERROR, "Unable to set nonblocking %d: %s", fd, strerror(errno));
370 goto error;
371 }
372 if (grpc_set_socket_cloexec(fd, 1) != GRPC_ERROR_NONE) {
373 gpr_log(GPR_ERROR, "Unable to set cloexec %d: %s", fd, strerror(errno));
374 goto error;
375 }
376
377 if (grpc_set_socket_ip_pktinfo_if_possible(fd) != GRPC_ERROR_NONE) {
378 gpr_log(GPR_ERROR, "Unable to set ip_pktinfo.");
379 goto error;
380 } else if (addr_ptr->sa_family == AF_INET6) {
381 if (grpc_set_socket_ipv6_recvpktinfo_if_possible(fd) != GRPC_ERROR_NONE) {
382 gpr_log(GPR_ERROR, "Unable to set ipv6_recvpktinfo.");
383 goto error;
384 }
385 }
386
387 if (grpc_set_socket_sndbuf(fd, snd_buf_size) != GRPC_ERROR_NONE) {
388 gpr_log(GPR_ERROR, "Failed to set send buffer size to %d bytes",
389 snd_buf_size);
390 goto error;
391 }
392
393 if (grpc_set_socket_rcvbuf(fd, rcv_buf_size) != GRPC_ERROR_NONE) {
394 gpr_log(GPR_ERROR, "Failed to set receive buffer size to %d bytes",
395 rcv_buf_size);
396 goto error;
397 }
398
399 {
400 int get_overflow = 1;
401 if (0 != setsockopt(fd, SOL_SOCKET, SO_RXQ_OVFL, &get_overflow,
402 sizeof(get_overflow))) {
403 gpr_log(GPR_INFO, "Failed to set socket overflow support");
404 }
405 }
406
407 if (so_reuseport && !grpc_is_unix_socket(addr) &&
408 grpc_set_socket_reuse_port(fd, 1) != GRPC_ERROR_NONE) {
409 gpr_log(GPR_ERROR, "Failed to set SO_REUSEPORT for fd %d", fd);
410 goto error;
411 }
412
413 if (bind_socket(socket_factory, fd, addr) < 0) {
414 char* addr_str;
415 grpc_sockaddr_to_string(&addr_str, addr, 0);
416 gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno));
417 gpr_free(addr_str);
418 goto error;
419 }
420
421 sockname_temp.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
422
423 if (getsockname(fd, reinterpret_cast<grpc_sockaddr*>(sockname_temp.addr),
424 &sockname_temp.len) < 0) {
425 gpr_log(GPR_ERROR, "Unable to get the address socket %d is bound to: %s",
426 fd, strerror(errno));
427 goto error;
428 }
429
430 return grpc_sockaddr_get_port(&sockname_temp);
431
432 error:
433 if (fd >= 0) {
434 close(fd);
435 }
436 return -1;
437 }
438
439 // static
do_read(void * arg,grpc_error * error)440 void GrpcUdpListener::do_read(void* arg, grpc_error* error) {
441 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
442 GPR_ASSERT(error == GRPC_ERROR_NONE);
443 /* TODO: the reason we hold server->mu here is merely to prevent fd
444 * shutdown while we are reading. However, it blocks do_write(). Switch to
445 * read lock if available. */
446 gpr_mu_lock(sp->mutex());
447 /* Tell the registered callback that data is available to read. */
448 if (!sp->already_shutdown_ && sp->udp_handler_->Read()) {
449 /* There maybe more packets to read. Schedule read_more_cb_ closure to run
450 * after finishing this event loop. */
451 GRPC_CLOSURE_SCHED(&sp->do_read_closure_, GRPC_ERROR_NONE);
452 } else {
453 /* Finish reading all the packets, re-arm the notification event so we can
454 * get another chance to read. Or fd already shutdown, re-arm to get a
455 * notification with shutdown error. */
456 grpc_fd_notify_on_read(sp->emfd_, &sp->read_closure_);
457 }
458 gpr_mu_unlock(sp->mutex());
459 }
460
461 // static
on_read(void * arg,grpc_error * error)462 void GrpcUdpListener::on_read(void* arg, grpc_error* error) {
463 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
464 sp->OnRead(error, arg);
465 }
466
OnRead(grpc_error * error,void * do_read_arg)467 void GrpcUdpListener::OnRead(grpc_error* error, void* do_read_arg) {
468 if (error != GRPC_ERROR_NONE) {
469 gpr_mu_lock(&server_->mu);
470 if (0 == --server_->active_ports && server_->shutdown) {
471 gpr_mu_unlock(&server_->mu);
472 deactivated_all_ports(server_);
473 } else {
474 gpr_mu_unlock(&server_->mu);
475 }
476 return;
477 }
478
479 /* Read once. If there is more data to read, off load the work to another
480 * thread to finish. */
481 if (udp_handler_->Read()) {
482 /* There maybe more packets to read. Schedule read_more_cb_ closure to run
483 * after finishing this event loop. */
484 GRPC_CLOSURE_INIT(
485 &do_read_closure_, do_read, do_read_arg,
486 grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::LONG));
487 GRPC_CLOSURE_SCHED(&do_read_closure_, GRPC_ERROR_NONE);
488 } else {
489 /* Finish reading all the packets, re-arm the notification event so we can
490 * get another chance to read. Or fd already shutdown, re-arm to get a
491 * notification with shutdown error. */
492 grpc_fd_notify_on_read(emfd_, &read_closure_);
493 }
494 }
495
496 // static
497 // Wrapper of grpc_fd_notify_on_write() with a grpc_closure callback interface.
fd_notify_on_write_wrapper(void * arg,grpc_error * error)498 void GrpcUdpListener::fd_notify_on_write_wrapper(void* arg, grpc_error* error) {
499 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
500 gpr_mu_lock(sp->mutex());
501 if (!sp->notify_on_write_armed_) {
502 grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
503 sp->notify_on_write_armed_ = true;
504 }
505 gpr_mu_unlock(sp->mutex());
506 }
507
508 // static
do_write(void * arg,grpc_error * error)509 void GrpcUdpListener::do_write(void* arg, grpc_error* error) {
510 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
511 gpr_mu_lock(sp->mutex());
512 if (sp->already_shutdown_) {
513 // If fd has been shutdown, don't write any more and re-arm notification.
514 grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
515 } else {
516 sp->notify_on_write_armed_ = false;
517 /* Tell the registered callback that the socket is writeable. */
518 GPR_ASSERT(error == GRPC_ERROR_NONE);
519 GRPC_CLOSURE_INIT(&sp->notify_on_write_closure_, fd_notify_on_write_wrapper,
520 arg, grpc_schedule_on_exec_ctx);
521 sp->udp_handler_->OnCanWrite(sp->server_->user_data,
522 &sp->notify_on_write_closure_);
523 }
524 gpr_mu_unlock(sp->mutex());
525 }
526
527 // static
on_write(void * arg,grpc_error * error)528 void GrpcUdpListener::on_write(void* arg, grpc_error* error) {
529 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
530 sp->OnCanWrite(error, arg);
531 }
532
OnCanWrite(grpc_error * error,void * do_write_arg)533 void GrpcUdpListener::OnCanWrite(grpc_error* error, void* do_write_arg) {
534 if (error != GRPC_ERROR_NONE) {
535 gpr_mu_lock(&server_->mu);
536 if (0 == --server_->active_ports && server_->shutdown) {
537 gpr_mu_unlock(&server_->mu);
538 deactivated_all_ports(server_);
539 } else {
540 gpr_mu_unlock(&server_->mu);
541 }
542 return;
543 }
544
545 /* Schedule actual write in another thread. */
546 GRPC_CLOSURE_INIT(
547 &do_write_closure_, do_write, do_write_arg,
548 grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::LONG));
549
550 GRPC_CLOSURE_SCHED(&do_write_closure_, GRPC_ERROR_NONE);
551 }
552
add_socket_to_server(grpc_udp_server * s,int fd,const grpc_resolved_address * addr,int rcv_buf_size,int snd_buf_size)553 static int add_socket_to_server(grpc_udp_server* s, int fd,
554 const grpc_resolved_address* addr,
555 int rcv_buf_size, int snd_buf_size) {
556 gpr_log(GPR_DEBUG, "add socket %d to server", fd);
557
558 int port = prepare_socket(s->socket_factory, fd, addr, rcv_buf_size,
559 snd_buf_size, s->so_reuseport);
560 if (port >= 0) {
561 gpr_mu_lock(&s->mu);
562 s->listeners.emplace_back(s, fd, addr);
563 gpr_log(GPR_DEBUG,
564 "add socket %d to server for port %d, %zu listener(s) in total", fd,
565 port, s->listeners.size());
566 gpr_mu_unlock(&s->mu);
567 }
568 return port;
569 }
570
grpc_udp_server_add_port(grpc_udp_server * s,const grpc_resolved_address * addr,int rcv_buf_size,int snd_buf_size,GrpcUdpHandlerFactory * handler_factory,size_t num_listeners)571 int grpc_udp_server_add_port(grpc_udp_server* s,
572 const grpc_resolved_address* addr,
573 int rcv_buf_size, int snd_buf_size,
574 GrpcUdpHandlerFactory* handler_factory,
575 size_t num_listeners) {
576 if (num_listeners > 1 && !s->so_reuseport) {
577 gpr_log(GPR_ERROR,
578 "Try to have multiple listeners on same port, but SO_REUSEPORT is "
579 "not supported. Only create 1 listener.");
580 }
581 char* addr_str;
582 grpc_sockaddr_to_string(&addr_str, addr, 1);
583 gpr_log(GPR_DEBUG, "add address: %s to server", addr_str);
584 gpr_free(addr_str);
585
586 int allocated_port1 = -1;
587 int allocated_port2 = -1;
588 int fd;
589 grpc_dualstack_mode dsmode;
590 grpc_resolved_address addr6_v4mapped;
591 grpc_resolved_address wild4;
592 grpc_resolved_address wild6;
593 grpc_resolved_address addr4_copy;
594 grpc_resolved_address* allocated_addr = nullptr;
595 grpc_resolved_address sockname_temp;
596 int port = 0;
597
598 /* Check if this is a wildcard port, and if so, try to keep the port the same
599 as some previously created listener. */
600 if (grpc_sockaddr_get_port(addr) == 0) {
601 /* Loop through existing listeners to find the port in use. */
602 for (size_t i = 0; i < s->listeners.size(); ++i) {
603 sockname_temp.len =
604 static_cast<socklen_t>(sizeof(struct sockaddr_storage));
605 if (0 == getsockname(s->listeners[i].fd(),
606 reinterpret_cast<grpc_sockaddr*>(sockname_temp.addr),
607 &sockname_temp.len)) {
608 port = grpc_sockaddr_get_port(&sockname_temp);
609 if (port > 0) {
610 /* Found such a port, update |addr| to reflects this port. */
611 allocated_addr = static_cast<grpc_resolved_address*>(
612 gpr_malloc(sizeof(grpc_resolved_address)));
613 memcpy(allocated_addr, addr, sizeof(grpc_resolved_address));
614 grpc_sockaddr_set_port(allocated_addr, port);
615 addr = allocated_addr;
616 break;
617 }
618 }
619 }
620 }
621
622 if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
623 addr = &addr6_v4mapped;
624 }
625
626 s->handler_factory = handler_factory;
627 for (size_t i = 0; i < num_listeners; ++i) {
628 /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
629 if (grpc_sockaddr_is_wildcard(addr, &port)) {
630 grpc_sockaddr_make_wildcards(port, &wild4, &wild6);
631
632 /* Try listening on IPv6 first. */
633 addr = &wild6;
634 // TODO(rjshade): Test and propagate the returned grpc_error*:
635 GRPC_ERROR_UNREF(grpc_create_dualstack_socket_using_factory(
636 s->socket_factory, addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd));
637 allocated_port1 =
638 add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
639 if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) {
640 if (port == 0) {
641 /* This is the first time to bind to |addr|. If its port is still
642 * wildcard port, update |addr| with the ephermeral port returned by
643 * kernel. Thus |addr| can have a specific port in following
644 * iterations. */
645 grpc_sockaddr_set_port(addr, allocated_port1);
646 port = allocated_port1;
647 } else if (allocated_port1 >= 0) {
648 /* The following successfully created socket should have same port as
649 * the first one. */
650 GPR_ASSERT(port == allocated_port1);
651 }
652 /* A dualstack socket is created, no need to create corresponding IPV4
653 * socket. */
654 continue;
655 }
656
657 /* If we didn't get a dualstack socket, also listen on 0.0.0.0. */
658 if (port == 0 && allocated_port1 > 0) {
659 /* |port| hasn't been assigned to an emphemeral port yet, |wild4| must
660 * have a wildcard port. Update it with the emphemeral port created
661 * during binding.*/
662 grpc_sockaddr_set_port(&wild4, allocated_port1);
663 port = allocated_port1;
664 }
665 /* |wild4| should have been updated with an emphemeral port by now. Use
666 * this IPV4 address to create a IPV4 socket. */
667 addr = &wild4;
668 }
669
670 // TODO(rjshade): Test and propagate the returned grpc_error*:
671 GRPC_ERROR_UNREF(grpc_create_dualstack_socket_using_factory(
672 s->socket_factory, addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd));
673 if (fd < 0) {
674 gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
675 }
676 if (dsmode == GRPC_DSMODE_IPV4 &&
677 grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
678 addr = &addr4_copy;
679 }
680 allocated_port2 =
681 add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
682 if (port == 0) {
683 /* Update |addr| with the ephermeral port returned by kernel. So |addr|
684 * can have a specific port in following iterations. */
685 grpc_sockaddr_set_port(addr, allocated_port2);
686 port = allocated_port2;
687 } else if (allocated_port2 >= 0) {
688 GPR_ASSERT(port == allocated_port2);
689 }
690 }
691
692 gpr_free(allocated_addr);
693 return port;
694 }
695
grpc_udp_server_get_fd(grpc_udp_server * s,unsigned port_index)696 int grpc_udp_server_get_fd(grpc_udp_server* s, unsigned port_index) {
697 if (port_index >= s->listeners.size()) {
698 return -1;
699 }
700
701 return s->listeners[port_index].fd();
702 }
703
grpc_udp_server_start(grpc_udp_server * s,grpc_pollset ** pollsets,size_t pollset_count,void * user_data)704 void grpc_udp_server_start(grpc_udp_server* s, grpc_pollset** pollsets,
705 size_t pollset_count, void* user_data) {
706 gpr_log(GPR_DEBUG, "grpc_udp_server_start");
707 gpr_mu_lock(&s->mu);
708 GPR_ASSERT(s->active_ports == 0);
709 s->pollsets = pollsets;
710 s->user_data = user_data;
711
712 for (size_t i = 0; i < s->listeners.size(); ++i) {
713 s->listeners[i].StartListening(pollsets, pollset_count, s->handler_factory);
714 }
715
716 gpr_mu_unlock(&s->mu);
717 }
718
StartListening(grpc_pollset ** pollsets,size_t pollset_count,GrpcUdpHandlerFactory * handler_factory)719 void GrpcUdpListener::StartListening(grpc_pollset** pollsets,
720 size_t pollset_count,
721 GrpcUdpHandlerFactory* handler_factory) {
722 gpr_mu_lock(&mutex_);
723 handler_factory_ = handler_factory;
724 udp_handler_ = handler_factory->CreateUdpHandler(emfd_, server_->user_data);
725 for (size_t i = 0; i < pollset_count; i++) {
726 grpc_pollset_add_fd(pollsets[i], emfd_);
727 }
728 GRPC_CLOSURE_INIT(&read_closure_, on_read, this, grpc_schedule_on_exec_ctx);
729 grpc_fd_notify_on_read(emfd_, &read_closure_);
730
731 GRPC_CLOSURE_INIT(&write_closure_, on_write, this, grpc_schedule_on_exec_ctx);
732 notify_on_write_armed_ = true;
733 grpc_fd_notify_on_write(emfd_, &write_closure_);
734
735 /* Registered for both read and write callbacks: increment active_ports
736 * twice to account for this, and delay free-ing of memory until both
737 * on_read and on_write have fired. */
738 server_->active_ports += 2;
739 gpr_mu_unlock(&mutex_);
740 }
741
OnDestroy()742 void GrpcUdpListener::OnDestroy() {
743 if (udp_handler_ != nullptr) {
744 handler_factory_->DestroyUdpHandler(udp_handler_);
745 }
746 }
747
748 #endif
749