3 * Copyright 2015 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
19 /* FIXME: "posix" files shouldn't be depending on _GNU_SOURCE */
25 #define SO_RXQ_OVFL 40
28 #include <grpc/support/port_platform.h>
30 #include "src/core/lib/iomgr/port.h"
32 #ifdef GRPC_POSIX_SOCKET
34 #include "src/core/lib/iomgr/udp_server.h"
39 #include <netinet/in.h>
40 #include <netinet/tcp.h>
42 #include <sys/socket.h>
44 #include <sys/types.h>
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"
67 /* A listener which implements basic features of Listening on a port for
69 class GrpcUdpListener {
71 GrpcUdpListener(grpc_udp_server* server, int fd,
72 const grpc_resolved_address* addr);
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);
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);
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);
87 /* Called when the grpc_fd is about to be orphaned (and the FD closed). */
88 void OnFdAboutToOrphan();
90 /* Called to orphan fd of this listener.*/
93 /* Called when this listener is going to be destroyed. */
96 int fd() const { return fd_; }
99 grpc_fd* emfd() const { return emfd_; }
101 gpr_mu* mutex() { return &mutex_; }
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);
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
112 static void fd_notify_on_write_wrapper(void* arg, grpc_error* error);
114 static void shutdown_fd(void* args, grpc_error* error);
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
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.
144 GrpcUdpListener::GrpcUdpListener(grpc_udp_server* server, int fd,
145 const grpc_resolved_address* addr)
148 orphan_notified_(false),
149 already_shutdown_(false) {
152 grpc_sockaddr_to_string(&addr_str, addr, 1);
153 gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
155 emfd_ = grpc_fd_create(fd, name, true);
156 memcpy(&addr_, addr, sizeof(grpc_resolved_address));
159 gpr_mu_init(&mutex_);
162 GrpcUdpListener::~GrpcUdpListener() { gpr_mu_destroy(&mutex_); }
164 /* the overall server */
165 struct grpc_udp_server {
168 /* factory to use for creating and binding sockets, or NULL */
169 grpc_socket_factory* socket_factory;
171 /* active port count: how many ports are actually still listening */
173 /* destroyed port count: how many ports are completely destroyed */
174 size_t destroyed_ports;
176 /* is this server shutting down? (boolean) */
179 /* An array of listeners */
180 grpc_core::InlinedVector<GrpcUdpListener, 16> listeners;
182 /* factory for use to create udp listeners */
183 GrpcUdpHandlerFactory* handler_factory;
185 /* shutdown callback */
186 grpc_closure* shutdown_complete;
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 */
195 /* latch has_so_reuseport during server creation */
199 static grpc_socket_factory* get_socket_factory(const grpc_channel_args* args) {
201 const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_SOCKET_FACTORY);
203 GPR_ASSERT(arg->type == GRPC_ARG_POINTER);
204 return static_cast<grpc_socket_factory*>(arg->value.pointer.p);
210 grpc_udp_server* grpc_udp_server_create(const grpc_channel_args* args) {
211 grpc_udp_server* s = grpc_core::New<grpc_udp_server>();
213 s->socket_factory = get_socket_factory(args);
214 if (s->socket_factory) {
215 grpc_socket_factory_ref(s->socket_factory);
218 s->destroyed_ports = 0;
220 s->so_reuseport = grpc_is_socket_reuse_port_supported();
225 void GrpcUdpListener::shutdown_fd(void* args, grpc_error* error) {
226 if (args == nullptr) {
227 // No-op if shutdown args are null.
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_);
241 gpr_mu_unlock(sp->mutex());
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);
249 gpr_mu_destroy(&s->mu);
251 gpr_log(GPR_DEBUG, "Destroy all listeners.");
252 for (size_t i = 0; i < s->listeners.size(); ++i) {
253 s->listeners[i].OnDestroy();
256 if (s->socket_factory) {
257 grpc_socket_factory_unref(s->socket_factory);
260 grpc_core::Delete(s);
263 static void destroyed_port(void* server, grpc_error* error) {
264 grpc_udp_server* s = static_cast<grpc_udp_server*>(server);
266 s->destroyed_ports++;
267 if (s->destroyed_ports == s->listeners.size()) {
268 gpr_mu_unlock(&s->mu);
271 gpr_mu_unlock(&s->mu);
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
278 static void deactivated_all_ports(grpc_udp_server* s) {
279 /* delete ALL the things */
282 GPR_ASSERT(s->shutdown);
284 if (s->listeners.size() == 0) {
285 gpr_mu_unlock(&s->mu);
289 for (size_t i = 0; i < s->listeners.size(); ++i) {
290 s->listeners[i].OrphanFd();
292 gpr_mu_unlock(&s->mu);
295 void GrpcUdpListener::OrphanFd() {
296 gpr_log(GPR_DEBUG, "Orphan fd %d, emfd %p", fd_, emfd_);
297 grpc_unlink_if_unix_domain_socket(&addr_);
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");
306 void grpc_udp_server_destroy(grpc_udp_server* s, grpc_closure* on_done) {
309 GPR_ASSERT(!s->shutdown);
312 s->shutdown_complete = on_done;
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();
321 gpr_mu_unlock(&s->mu);
323 gpr_mu_unlock(&s->mu);
324 deactivated_all_ports(s);
328 void GrpcUdpListener::OnFdAboutToOrphan() {
329 gpr_mu_lock(&mutex_);
330 grpc_unlink_if_unix_domain_socket(&addr_);
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;
343 gpr_mu_unlock(&mutex_);
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)
351 reinterpret_cast<grpc_sockaddr*>(
352 const_cast<char*>(addr->addr)),
356 /* Prepare a recently-created socket for listening. */
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));
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));
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));
377 if (grpc_set_socket_ip_pktinfo_if_possible(fd) != GRPC_ERROR_NONE) {
378 gpr_log(GPR_ERROR, "Unable to set ip_pktinfo.");
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.");
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",
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",
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");
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);
413 if (bind_socket(socket_factory, fd, addr) < 0) {
415 grpc_sockaddr_to_string(&addr_str, addr, 0);
416 gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno));
421 sockname_temp.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
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));
430 return grpc_sockaddr_get_port(&sockname_temp);
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);
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_);
458 gpr_mu_unlock(sp->mutex());
462 void GrpcUdpListener::on_read(void* arg, grpc_error* error) {
463 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
464 sp->OnRead(error, 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_);
474 gpr_mu_unlock(&server_->mu);
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. */
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);
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_);
497 // Wrapper of grpc_fd_notify_on_write() with a grpc_closure callback interface.
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;
505 gpr_mu_unlock(sp->mutex());
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_);
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_);
524 gpr_mu_unlock(sp->mutex());
528 void GrpcUdpListener::on_write(void* arg, grpc_error* error) {
529 GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
530 sp->OnCanWrite(error, 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_);
540 gpr_mu_unlock(&server_->mu);
545 /* Schedule actual write in another thread. */
547 &do_write_closure_, do_write, do_write_arg,
548 grpc_core::Executor::Scheduler(grpc_core::ExecutorJobType::LONG));
550 GRPC_CLOSURE_SCHED(&do_write_closure_, GRPC_ERROR_NONE);
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);
558 int port = prepare_socket(s->socket_factory, fd, addr, rcv_buf_size,
559 snd_buf_size, s->so_reuseport);
562 s->listeners.emplace_back(s, fd, addr);
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);
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) {
578 "Try to have multiple listeners on same port, but SO_REUSEPORT is "
579 "not supported. Only create 1 listener.");
582 grpc_sockaddr_to_string(&addr_str, addr, 1);
583 gpr_log(GPR_DEBUG, "add address: %s to server", addr_str);
586 int allocated_port1 = -1;
587 int allocated_port2 = -1;
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;
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) {
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);
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;
622 if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
623 addr = &addr6_v4mapped;
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);
632 /* Try listening on IPv6 first. */
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));
638 add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
639 if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) {
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
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
650 GPR_ASSERT(port == allocated_port1);
652 /* A dualstack socket is created, no need to create corresponding IPV4
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
662 grpc_sockaddr_set_port(&wild4, allocated_port1);
663 port = allocated_port1;
665 /* |wild4| should have been updated with an emphemeral port by now. Use
666 * this IPV4 address to create a IPV4 socket. */
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));
674 gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
676 if (dsmode == GRPC_DSMODE_IPV4 &&
677 grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
681 add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
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);
692 gpr_free(allocated_addr);
696 int grpc_udp_server_get_fd(grpc_udp_server* s, unsigned port_index) {
697 if (port_index >= s->listeners.size()) {
701 return s->listeners[port_index].fd();
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");
708 GPR_ASSERT(s->active_ports == 0);
709 s->pollsets = pollsets;
710 s->user_data = user_data;
712 for (size_t i = 0; i < s->listeners.size(); ++i) {
713 s->listeners[i].StartListening(pollsets, pollset_count, s->handler_factory);
716 gpr_mu_unlock(&s->mu);
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_);
728 GRPC_CLOSURE_INIT(&read_closure_, on_read, this, grpc_schedule_on_exec_ctx);
729 grpc_fd_notify_on_read(emfd_, &read_closure_);
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_);
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_);
742 void GrpcUdpListener::OnDestroy() {
743 if (udp_handler_ != nullptr) {
744 handler_factory_->DestroyUdpHandler(udp_handler_);