blob: 3dd7cab855ced9efc2a7f96e6efef50c2eb8b088 [file] [log] [blame]
/*
*
* Copyright 2015 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
/* FIXME: "posix" files shouldn't be depending on _GNU_SOURCE */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#ifndef SO_RXQ_OVFL
#define SO_RXQ_OVFL 40
#endif
#include <grpc/support/port_platform.h>
#include "src/core/lib/iomgr/port.h"
#ifdef GRPC_POSIX_SOCKET
#include "src/core/lib/iomgr/udp_server.h"
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include <grpc/support/time.h>
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/gprpp/inlined_vector.h"
#include "src/core/lib/gprpp/memory.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/ev_posix.h"
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/lib/iomgr/sockaddr.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
#include "src/core/lib/iomgr/socket_factory_posix.h"
#include "src/core/lib/iomgr/socket_utils_posix.h"
#include "src/core/lib/iomgr/unix_sockets_posix.h"
/* A listener which implements basic features of Listening on a port for
* I/O events*/
class GrpcUdpListener {
public:
GrpcUdpListener(grpc_udp_server* server, int fd,
const grpc_resolved_address* addr);
~GrpcUdpListener();
/* Called when grpc server starts to listening on the grpc_fd. */
void StartListening(grpc_pollset** pollsets, size_t pollset_count,
GrpcUdpHandlerFactory* handler_factory);
/* Called when data is available to read from the socket.
* Return true if there is more data to read from fd. */
void OnRead(grpc_error* error, void* do_read_arg);
/* Called when the socket is writeable. The given closure should be scheduled
* when the socket becomes blocked next time. */
void OnCanWrite(grpc_error* error, void* do_write_arg);
/* Called when the grpc_fd is about to be orphaned (and the FD closed). */
void OnFdAboutToOrphan();
/* Called to orphan fd of this listener.*/
void OrphanFd();
/* Called when this listener is going to be destroyed. */
void OnDestroy();
int fd() const { return fd_; }
protected:
grpc_fd* emfd() const { return emfd_; }
gpr_mu* mutex() { return &mutex_; }
private:
/* event manager callback when reads are ready */
static void on_read(void* arg, grpc_error* error);
static void on_write(void* arg, grpc_error* error);
static void do_read(void* arg, grpc_error* error);
static void do_write(void* arg, grpc_error* error);
// Wrapper of grpc_fd_notify_on_write() with a grpc_closure callback
// interface.
static void fd_notify_on_write_wrapper(void* arg, grpc_error* error);
static void shutdown_fd(void* args, grpc_error* error);
int fd_;
grpc_fd* emfd_;
grpc_udp_server* server_;
grpc_resolved_address addr_;
grpc_closure read_closure_;
grpc_closure write_closure_;
// To be called when corresponding QuicGrpcServer closes all active
// connections.
grpc_closure orphan_fd_closure_;
grpc_closure destroyed_closure_;
// To be scheduled on another thread to actually read/write.
grpc_closure do_read_closure_;
grpc_closure do_write_closure_;
grpc_closure notify_on_write_closure_;
// True if orphan_cb is trigered.
bool orphan_notified_;
// True if grpc_fd_notify_on_write() is called after on_write() call.
bool notify_on_write_armed_;
// True if fd has been shutdown.
bool already_shutdown_;
// Object actually handles I/O events. Assigned in StartListening().
GrpcUdpHandler* udp_handler_ = nullptr;
// To be notified on destruction.
GrpcUdpHandlerFactory* handler_factory_ = nullptr;
// Required to access above fields.
gpr_mu mutex_;
};
GrpcUdpListener::GrpcUdpListener(grpc_udp_server* server, int fd,
const grpc_resolved_address* addr)
: fd_(fd),
server_(server),
orphan_notified_(false),
already_shutdown_(false) {
char* addr_str;
char* name;
grpc_sockaddr_to_string(&addr_str, addr, 1);
gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
gpr_free(addr_str);
emfd_ = grpc_fd_create(fd, name, true);
memcpy(&addr_, addr, sizeof(grpc_resolved_address));
GPR_ASSERT(emfd_);
gpr_free(name);
gpr_mu_init(&mutex_);
}
GrpcUdpListener::~GrpcUdpListener() { gpr_mu_destroy(&mutex_); }
/* the overall server */
struct grpc_udp_server {
gpr_mu mu;
/* factory to use for creating and binding sockets, or NULL */
grpc_socket_factory* socket_factory;
/* active port count: how many ports are actually still listening */
size_t active_ports;
/* destroyed port count: how many ports are completely destroyed */
size_t destroyed_ports;
/* is this server shutting down? (boolean) */
int shutdown;
/* An array of listeners */
grpc_core::InlinedVector<GrpcUdpListener, 16> listeners;
/* factory for use to create udp listeners */
GrpcUdpHandlerFactory* handler_factory;
/* shutdown callback */
grpc_closure* shutdown_complete;
/* all pollsets interested in new connections */
grpc_pollset** pollsets;
/* number of pollsets in the pollsets array */
size_t pollset_count;
/* opaque object to pass to callbacks */
void* user_data;
/* latch has_so_reuseport during server creation */
bool so_reuseport;
};
static grpc_socket_factory* get_socket_factory(const grpc_channel_args* args) {
if (args) {
const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_SOCKET_FACTORY);
if (arg) {
GPR_ASSERT(arg->type == GRPC_ARG_POINTER);
return static_cast<grpc_socket_factory*>(arg->value.pointer.p);
}
}
return nullptr;
}
grpc_udp_server* grpc_udp_server_create(const grpc_channel_args* args) {
grpc_udp_server* s = grpc_core::New<grpc_udp_server>();
gpr_mu_init(&s->mu);
s->socket_factory = get_socket_factory(args);
if (s->socket_factory) {
grpc_socket_factory_ref(s->socket_factory);
}
s->active_ports = 0;
s->destroyed_ports = 0;
s->shutdown = 0;
s->so_reuseport = grpc_is_socket_reuse_port_supported();
return s;
}
// static
void GrpcUdpListener::shutdown_fd(void* args, grpc_error* error) {
if (args == nullptr) {
// No-op if shutdown args are null.
return;
}
auto sp = static_cast<GrpcUdpListener*>(args);
gpr_mu_lock(sp->mutex());
gpr_log(GPR_DEBUG, "shutdown fd %d", sp->fd_);
grpc_fd_shutdown(sp->emfd_, GRPC_ERROR_REF(error));
sp->already_shutdown_ = true;
if (!sp->notify_on_write_armed_) {
// Re-arm write notification to notify listener with error. This is
// necessary to decrement active_ports.
sp->notify_on_write_armed_ = true;
grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
}
gpr_mu_unlock(sp->mutex());
}
static void finish_shutdown(grpc_udp_server* s) {
if (s->shutdown_complete != nullptr) {
GRPC_CLOSURE_SCHED(s->shutdown_complete, GRPC_ERROR_NONE);
}
gpr_mu_destroy(&s->mu);
gpr_log(GPR_DEBUG, "Destroy all listeners.");
for (size_t i = 0; i < s->listeners.size(); ++i) {
s->listeners[i].OnDestroy();
}
if (s->socket_factory) {
grpc_socket_factory_unref(s->socket_factory);
}
grpc_core::Delete(s);
}
static void destroyed_port(void* server, grpc_error* error) {
grpc_udp_server* s = static_cast<grpc_udp_server*>(server);
gpr_mu_lock(&s->mu);
s->destroyed_ports++;
if (s->destroyed_ports == s->listeners.size()) {
gpr_mu_unlock(&s->mu);
finish_shutdown(s);
} else {
gpr_mu_unlock(&s->mu);
}
}
/* called when all listening endpoints have been shutdown, so no further
events will be received on them - at this point it's safe to destroy
things */
static void deactivated_all_ports(grpc_udp_server* s) {
/* delete ALL the things */
gpr_mu_lock(&s->mu);
GPR_ASSERT(s->shutdown);
if (s->listeners.size() == 0) {
gpr_mu_unlock(&s->mu);
finish_shutdown(s);
return;
}
for (size_t i = 0; i < s->listeners.size(); ++i) {
s->listeners[i].OrphanFd();
}
gpr_mu_unlock(&s->mu);
}
void GrpcUdpListener::OrphanFd() {
gpr_log(GPR_DEBUG, "Orphan fd %d, emfd %p", fd_, emfd_);
grpc_unlink_if_unix_domain_socket(&addr_);
GRPC_CLOSURE_INIT(&destroyed_closure_, destroyed_port, server_,
grpc_schedule_on_exec_ctx);
/* Because at this point, all listening sockets have been shutdown already, no
* need to call OnFdAboutToOrphan() to notify the handler again. */
grpc_fd_orphan(emfd_, &destroyed_closure_, nullptr, "udp_listener_shutdown");
}
void grpc_udp_server_destroy(grpc_udp_server* s, grpc_closure* on_done) {
gpr_mu_lock(&s->mu);
GPR_ASSERT(!s->shutdown);
s->shutdown = 1;
s->shutdown_complete = on_done;
gpr_log(GPR_DEBUG, "start to destroy udp_server");
/* shutdown all fd's */
if (s->active_ports) {
for (size_t i = 0; i < s->listeners.size(); ++i) {
GrpcUdpListener* sp = &s->listeners[i];
sp->OnFdAboutToOrphan();
}
gpr_mu_unlock(&s->mu);
} else {
gpr_mu_unlock(&s->mu);
deactivated_all_ports(s);
}
}
void GrpcUdpListener::OnFdAboutToOrphan() {
gpr_mu_lock(&mutex_);
grpc_unlink_if_unix_domain_socket(&addr_);
GRPC_CLOSURE_INIT(&destroyed_closure_, destroyed_port, server_,
grpc_schedule_on_exec_ctx);
if (!orphan_notified_ && udp_handler_ != nullptr) {
/* Singals udp_handler that the FD is about to be closed and
* should no longer be used. */
GRPC_CLOSURE_INIT(&orphan_fd_closure_, shutdown_fd, this,
grpc_schedule_on_exec_ctx);
gpr_log(GPR_DEBUG, "fd %d about to be orphaned", fd_);
udp_handler_->OnFdAboutToOrphan(&orphan_fd_closure_, server_->user_data);
orphan_notified_ = true;
}
gpr_mu_unlock(&mutex_);
}
static int bind_socket(grpc_socket_factory* socket_factory, int sockfd,
const grpc_resolved_address* addr) {
return (socket_factory != nullptr)
? grpc_socket_factory_bind(socket_factory, sockfd, addr)
: bind(sockfd,
reinterpret_cast<grpc_sockaddr*>(
const_cast<char*>(addr->addr)),
addr->len);
}
/* Prepare a recently-created socket for listening. */
static int 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) {
grpc_resolved_address sockname_temp;
grpc_sockaddr* addr_ptr =
reinterpret_cast<grpc_sockaddr*>(const_cast<char*>(addr->addr));
if (fd < 0) {
goto error;
}
if (grpc_set_socket_nonblocking(fd, 1) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Unable to set nonblocking %d: %s", fd, strerror(errno));
goto error;
}
if (grpc_set_socket_cloexec(fd, 1) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Unable to set cloexec %d: %s", fd, strerror(errno));
goto error;
}
if (grpc_set_socket_ip_pktinfo_if_possible(fd) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Unable to set ip_pktinfo.");
goto error;
} else if (addr_ptr->sa_family == AF_INET6) {
if (grpc_set_socket_ipv6_recvpktinfo_if_possible(fd) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Unable to set ipv6_recvpktinfo.");
goto error;
}
}
if (grpc_set_socket_sndbuf(fd, snd_buf_size) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Failed to set send buffer size to %d bytes",
snd_buf_size);
goto error;
}
if (grpc_set_socket_rcvbuf(fd, rcv_buf_size) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Failed to set receive buffer size to %d bytes",
rcv_buf_size);
goto error;
}
{
int get_overflow = 1;
if (0 != setsockopt(fd, SOL_SOCKET, SO_RXQ_OVFL, &get_overflow,
sizeof(get_overflow))) {
gpr_log(GPR_INFO, "Failed to set socket overflow support");
}
}
if (so_reuseport && !grpc_is_unix_socket(addr) &&
grpc_set_socket_reuse_port(fd, 1) != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Failed to set SO_REUSEPORT for fd %d", fd);
goto error;
}
if (bind_socket(socket_factory, fd, addr) < 0) {
char* addr_str;
grpc_sockaddr_to_string(&addr_str, addr, 0);
gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno));
gpr_free(addr_str);
goto error;
}
sockname_temp.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
if (getsockname(fd, reinterpret_cast<grpc_sockaddr*>(sockname_temp.addr),
&sockname_temp.len) < 0) {
gpr_log(GPR_ERROR, "Unable to get the address socket %d is bound to: %s",
fd, strerror(errno));
goto error;
}
return grpc_sockaddr_get_port(&sockname_temp);
error:
if (fd >= 0) {
close(fd);
}
return -1;
}
// static
void GrpcUdpListener::do_read(void* arg, grpc_error* error) {
GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
GPR_ASSERT(error == GRPC_ERROR_NONE);
/* TODO: the reason we hold server->mu here is merely to prevent fd
* shutdown while we are reading. However, it blocks do_write(). Switch to
* read lock if available. */
gpr_mu_lock(sp->mutex());
/* Tell the registered callback that data is available to read. */
if (!sp->already_shutdown_ && sp->udp_handler_->Read()) {
/* There maybe more packets to read. Schedule read_more_cb_ closure to run
* after finishing this event loop. */
GRPC_CLOSURE_SCHED(&sp->do_read_closure_, GRPC_ERROR_NONE);
} else {
/* Finish reading all the packets, re-arm the notification event so we can
* get another chance to read. Or fd already shutdown, re-arm to get a
* notification with shutdown error. */
grpc_fd_notify_on_read(sp->emfd_, &sp->read_closure_);
}
gpr_mu_unlock(sp->mutex());
}
// static
void GrpcUdpListener::on_read(void* arg, grpc_error* error) {
GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
sp->OnRead(error, arg);
}
void GrpcUdpListener::OnRead(grpc_error* error, void* do_read_arg) {
if (error != GRPC_ERROR_NONE) {
gpr_mu_lock(&server_->mu);
if (0 == --server_->active_ports && server_->shutdown) {
gpr_mu_unlock(&server_->mu);
deactivated_all_ports(server_);
} else {
gpr_mu_unlock(&server_->mu);
}
return;
}
/* Read once. If there is more data to read, off load the work to another
* thread to finish. */
if (udp_handler_->Read()) {
/* There maybe more packets to read. Schedule read_more_cb_ closure to run
* after finishing this event loop. */
GRPC_CLOSURE_INIT(&do_read_closure_, do_read, do_read_arg,
grpc_executor_scheduler(GRPC_EXECUTOR_LONG));
GRPC_CLOSURE_SCHED(&do_read_closure_, GRPC_ERROR_NONE);
} else {
/* Finish reading all the packets, re-arm the notification event so we can
* get another chance to read. Or fd already shutdown, re-arm to get a
* notification with shutdown error. */
grpc_fd_notify_on_read(emfd_, &read_closure_);
}
}
// static
// Wrapper of grpc_fd_notify_on_write() with a grpc_closure callback interface.
void GrpcUdpListener::fd_notify_on_write_wrapper(void* arg, grpc_error* error) {
GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
gpr_mu_lock(sp->mutex());
if (!sp->notify_on_write_armed_) {
grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
sp->notify_on_write_armed_ = true;
}
gpr_mu_unlock(sp->mutex());
}
// static
void GrpcUdpListener::do_write(void* arg, grpc_error* error) {
GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
gpr_mu_lock(sp->mutex());
if (sp->already_shutdown_) {
// If fd has been shutdown, don't write any more and re-arm notification.
grpc_fd_notify_on_write(sp->emfd_, &sp->write_closure_);
} else {
sp->notify_on_write_armed_ = false;
/* Tell the registered callback that the socket is writeable. */
GPR_ASSERT(error == GRPC_ERROR_NONE);
GRPC_CLOSURE_INIT(&sp->notify_on_write_closure_, fd_notify_on_write_wrapper,
arg, grpc_schedule_on_exec_ctx);
sp->udp_handler_->OnCanWrite(sp->server_->user_data,
&sp->notify_on_write_closure_);
}
gpr_mu_unlock(sp->mutex());
}
// static
void GrpcUdpListener::on_write(void* arg, grpc_error* error) {
GrpcUdpListener* sp = static_cast<GrpcUdpListener*>(arg);
sp->OnCanWrite(error, arg);
}
void GrpcUdpListener::OnCanWrite(grpc_error* error, void* do_write_arg) {
if (error != GRPC_ERROR_NONE) {
gpr_mu_lock(&server_->mu);
if (0 == --server_->active_ports && server_->shutdown) {
gpr_mu_unlock(&server_->mu);
deactivated_all_ports(server_);
} else {
gpr_mu_unlock(&server_->mu);
}
return;
}
/* Schedule actual write in another thread. */
GRPC_CLOSURE_INIT(&do_write_closure_, do_write, do_write_arg,
grpc_executor_scheduler(GRPC_EXECUTOR_LONG));
GRPC_CLOSURE_SCHED(&do_write_closure_, GRPC_ERROR_NONE);
}
static int add_socket_to_server(grpc_udp_server* s, int fd,
const grpc_resolved_address* addr,
int rcv_buf_size, int snd_buf_size) {
gpr_log(GPR_DEBUG, "add socket %d to server", fd);
int port = prepare_socket(s->socket_factory, fd, addr, rcv_buf_size,
snd_buf_size, s->so_reuseport);
if (port >= 0) {
gpr_mu_lock(&s->mu);
s->listeners.emplace_back(s, fd, addr);
gpr_log(GPR_DEBUG,
"add socket %d to server for port %d, %zu listener(s) in total", fd,
port, s->listeners.size());
gpr_mu_unlock(&s->mu);
}
return port;
}
int 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) {
if (num_listeners > 1 && !s->so_reuseport) {
gpr_log(GPR_ERROR,
"Try to have multiple listeners on same port, but SO_REUSEPORT is "
"not supported. Only create 1 listener.");
}
char* addr_str;
grpc_sockaddr_to_string(&addr_str, addr, 1);
gpr_log(GPR_DEBUG, "add address: %s to server", addr_str);
gpr_free(addr_str);
int allocated_port1 = -1;
int allocated_port2 = -1;
int fd;
grpc_dualstack_mode dsmode;
grpc_resolved_address addr6_v4mapped;
grpc_resolved_address wild4;
grpc_resolved_address wild6;
grpc_resolved_address addr4_copy;
grpc_resolved_address* allocated_addr = nullptr;
grpc_resolved_address sockname_temp;
int port = 0;
/* Check if this is a wildcard port, and if so, try to keep the port the same
as some previously created listener. */
if (grpc_sockaddr_get_port(addr) == 0) {
/* Loop through existing listeners to find the port in use. */
for (size_t i = 0; i < s->listeners.size(); ++i) {
sockname_temp.len =
static_cast<socklen_t>(sizeof(struct sockaddr_storage));
if (0 == getsockname(s->listeners[i].fd(),
reinterpret_cast<grpc_sockaddr*>(sockname_temp.addr),
&sockname_temp.len)) {
port = grpc_sockaddr_get_port(&sockname_temp);
if (port > 0) {
/* Found such a port, update |addr| to reflects this port. */
allocated_addr = static_cast<grpc_resolved_address*>(
gpr_malloc(sizeof(grpc_resolved_address)));
memcpy(allocated_addr, addr, sizeof(grpc_resolved_address));
grpc_sockaddr_set_port(allocated_addr, port);
addr = allocated_addr;
break;
}
}
}
}
if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
addr = &addr6_v4mapped;
}
s->handler_factory = handler_factory;
for (size_t i = 0; i < num_listeners; ++i) {
/* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
if (grpc_sockaddr_is_wildcard(addr, &port)) {
grpc_sockaddr_make_wildcards(port, &wild4, &wild6);
/* Try listening on IPv6 first. */
addr = &wild6;
// TODO(rjshade): Test and propagate the returned grpc_error*:
GRPC_ERROR_UNREF(grpc_create_dualstack_socket_using_factory(
s->socket_factory, addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd));
allocated_port1 =
add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) {
if (port == 0) {
/* This is the first time to bind to |addr|. If its port is still
* wildcard port, update |addr| with the ephermeral port returned by
* kernel. Thus |addr| can have a specific port in following
* iterations. */
grpc_sockaddr_set_port(addr, allocated_port1);
port = allocated_port1;
} else if (allocated_port1 >= 0) {
/* The following sucessfully created socket should have same port as
* the first one. */
GPR_ASSERT(port == allocated_port1);
}
/* A dualstack socket is created, no need to create corresponding IPV4
* socket. */
continue;
}
/* If we didn't get a dualstack socket, also listen on 0.0.0.0. */
if (port == 0 && allocated_port1 > 0) {
/* |port| hasn't been assigned to an emphemeral port yet, |wild4| must
* have a wildcard port. Update it with the emphemeral port created
* during binding.*/
grpc_sockaddr_set_port(&wild4, allocated_port1);
port = allocated_port1;
}
/* |wild4| should have been updated with an emphemeral port by now. Use
* this IPV4 address to create a IPV4 socket. */
addr = &wild4;
}
// TODO(rjshade): Test and propagate the returned grpc_error*:
GRPC_ERROR_UNREF(grpc_create_dualstack_socket_using_factory(
s->socket_factory, addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd));
if (fd < 0) {
gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
}
if (dsmode == GRPC_DSMODE_IPV4 &&
grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
addr = &addr4_copy;
}
allocated_port2 =
add_socket_to_server(s, fd, addr, rcv_buf_size, snd_buf_size);
if (port == 0) {
/* Update |addr| with the ephermeral port returned by kernel. So |addr|
* can have a specific port in following iterations. */
grpc_sockaddr_set_port(addr, allocated_port2);
port = allocated_port2;
} else if (allocated_port2 >= 0) {
GPR_ASSERT(port == allocated_port2);
}
}
gpr_free(allocated_addr);
return port;
}
int grpc_udp_server_get_fd(grpc_udp_server* s, unsigned port_index) {
if (port_index >= s->listeners.size()) {
return -1;
}
return s->listeners[port_index].fd();
}
void grpc_udp_server_start(grpc_udp_server* s, grpc_pollset** pollsets,
size_t pollset_count, void* user_data) {
gpr_log(GPR_DEBUG, "grpc_udp_server_start");
gpr_mu_lock(&s->mu);
GPR_ASSERT(s->active_ports == 0);
s->pollsets = pollsets;
s->user_data = user_data;
for (size_t i = 0; i < s->listeners.size(); ++i) {
s->listeners[i].StartListening(pollsets, pollset_count, s->handler_factory);
}
gpr_mu_unlock(&s->mu);
}
void GrpcUdpListener::StartListening(grpc_pollset** pollsets,
size_t pollset_count,
GrpcUdpHandlerFactory* handler_factory) {
gpr_mu_lock(&mutex_);
handler_factory_ = handler_factory;
udp_handler_ = handler_factory->CreateUdpHandler(emfd_, server_->user_data);
for (size_t i = 0; i < pollset_count; i++) {
grpc_pollset_add_fd(pollsets[i], emfd_);
}
GRPC_CLOSURE_INIT(&read_closure_, on_read, this, grpc_schedule_on_exec_ctx);
grpc_fd_notify_on_read(emfd_, &read_closure_);
GRPC_CLOSURE_INIT(&write_closure_, on_write, this, grpc_schedule_on_exec_ctx);
notify_on_write_armed_ = true;
grpc_fd_notify_on_write(emfd_, &write_closure_);
/* Registered for both read and write callbacks: increment active_ports
* twice to account for this, and delay free-ing of memory until both
* on_read and on_write have fired. */
server_->active_ports += 2;
gpr_mu_unlock(&mutex_);
}
void GrpcUdpListener::OnDestroy() {
if (udp_handler_ != nullptr) {
handler_factory_->DestroyUdpHandler(udp_handler_);
}
}
#endif