diff options
Diffstat (limited to 'src/core/lib/iomgr/udp_server.c')
-rw-r--r-- | src/core/lib/iomgr/udp_server.c | 202 |
1 files changed, 106 insertions, 96 deletions
diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index edf7b133e9..fd0c7a0f9d 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -36,9 +36,9 @@ #define _GNU_SOURCE #endif -#include <grpc/support/port_platform.h> +#include "src/core/lib/iomgr/port.h" -#ifdef GPR_POSIX_SOCKET +#ifdef GRPC_POSIX_SOCKET #include "src/core/lib/iomgr/udp_server.h" @@ -62,32 +62,30 @@ #include "src/core/lib/iomgr/error.h" #include "src/core/lib/iomgr/ev_posix.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_utils_posix.h" +#include "src/core/lib/iomgr/unix_sockets_posix.h" #include "src/core/lib/support/string.h" -#define INIT_PORT_CAP 2 - /* one listening port */ -typedef struct { +typedef struct grpc_udp_listener grpc_udp_listener; +struct grpc_udp_listener { int fd; grpc_fd *emfd; grpc_udp_server *server; - union { - uint8_t untyped[GRPC_MAX_SOCKADDR_SIZE]; - struct sockaddr sockaddr; - } addr; - size_t addr_len; + grpc_resolved_address addr; grpc_closure read_closure; grpc_closure destroyed_closure; grpc_udp_server_read_cb read_cb; grpc_udp_server_orphan_cb orphan_cb; -} server_port; + + struct grpc_udp_listener *next; +}; /* the overall server */ struct grpc_udp_server { gpr_mu mu; - gpr_cv cv; /* active port count: how many ports are actually still listening */ size_t active_ports; @@ -97,10 +95,10 @@ struct grpc_udp_server { /* is this server shutting down? (boolean) */ int shutdown; - /* all listening ports */ - server_port *ports; - size_t nports; - size_t port_capacity; + /* linked list of server ports */ + grpc_udp_listener *head; + grpc_udp_listener *tail; + unsigned nports; /* shutdown callback */ grpc_closure *shutdown_complete; @@ -116,24 +114,29 @@ struct grpc_udp_server { grpc_udp_server *grpc_udp_server_create(void) { grpc_udp_server *s = gpr_malloc(sizeof(grpc_udp_server)); gpr_mu_init(&s->mu); - gpr_cv_init(&s->cv); s->active_ports = 0; s->destroyed_ports = 0; s->shutdown = 0; - s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); + s->head = NULL; + s->tail = NULL; s->nports = 0; - s->port_capacity = INIT_PORT_CAP; return s; } static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { - grpc_exec_ctx_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL); + if (s->shutdown_complete != NULL) { + grpc_exec_ctx_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL); + } gpr_mu_destroy(&s->mu); - gpr_cv_destroy(&s->cv); - gpr_free(s->ports); + while (s->head) { + grpc_udp_listener *sp = s->head; + s->head = sp->next; + gpr_free(sp); + } + gpr_free(s); } @@ -154,8 +157,6 @@ static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server, events will be received on them - at this point it's safe to destroy things */ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { - size_t i; - /* delete ALL the things */ gpr_mu_lock(&s->mu); @@ -164,9 +165,11 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { return; } - if (s->nports) { - for (i = 0; i < s->nports; i++) { - server_port *sp = &s->ports[i]; + if (s->head) { + grpc_udp_listener *sp; + for (sp = s->head; sp; sp = sp->next) { + grpc_unlink_if_unix_domain_socket(&sp->addr); + sp->destroyed_closure.cb = destroyed_port; sp->destroyed_closure.cb_arg = s; @@ -187,7 +190,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, grpc_closure *on_done) { - size_t i; + grpc_udp_listener *sp; gpr_mu_lock(&s->mu); GPR_ASSERT(!s->shutdown); @@ -197,14 +200,10 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, /* shutdown all fd's */ if (s->active_ports) { - for (i = 0; i < s->nports; i++) { - server_port *sp = &s->ports[i]; - /* Call the orphan_cb to signal that the FD is about to be closed and - * should no longer be used. */ + for (sp = s->head; sp; sp = sp->next) { GPR_ASSERT(sp->orphan_cb); sp->orphan_cb(sp->emfd); - - grpc_fd_shutdown(exec_ctx, s->ports[i].emfd); + grpc_fd_shutdown(exec_ctx, sp->emfd); } gpr_mu_unlock(&s->mu); } else { @@ -214,10 +213,9 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, } /* Prepare a recently-created socket for listening. */ -static int prepare_socket(int fd, const struct sockaddr *addr, - size_t addr_len) { - struct sockaddr_storage sockname_temp; - socklen_t sockname_len; +static int prepare_socket(int fd, const grpc_resolved_address *addr) { + grpc_resolved_address sockname_temp; + struct sockaddr *addr_ptr = (struct sockaddr *)addr->addr; /* Set send/receive socket buffers to 1 MB */ int buffer_size_bytes = 1024 * 1024; @@ -237,15 +235,15 @@ static int prepare_socket(int fd, const struct sockaddr *addr, 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->sa_family == AF_INET6) { + } 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; } } - GPR_ASSERT(addr_len < ~(socklen_t)0); - if (bind(fd, addr, (socklen_t)addr_len) < 0) { + GPR_ASSERT(addr->len < ~(socklen_t)0); + if (bind(fd, (struct sockaddr *)addr, (socklen_t)addr->len) < 0) { char *addr_str; grpc_sockaddr_to_string(&addr_str, addr, 0); gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno)); @@ -253,8 +251,10 @@ static int prepare_socket(int fd, const struct sockaddr *addr, goto error; } - sockname_len = sizeof(sockname_temp); - if (getsockname(fd, (struct sockaddr *)&sockname_temp, &sockname_len) < 0) { + sockname_temp.len = sizeof(struct sockaddr_storage); + + if (getsockname(fd, (struct sockaddr *)sockname_temp.addr, + (socklen_t *)&sockname_temp.len) < 0) { goto error; } @@ -270,7 +270,7 @@ static int prepare_socket(int fd, const struct sockaddr *addr, goto error; } - return grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + return grpc_sockaddr_get_port(&sockname_temp); error: if (fd >= 0) { @@ -281,10 +281,10 @@ error: /* event manager callback when reads are ready */ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - server_port *sp = arg; + grpc_udp_listener *sp = arg; + gpr_mu_lock(&sp->server->mu); if (error != GRPC_ERROR_NONE) { - gpr_mu_lock(&sp->server->mu); if (0 == --sp->server->active_ports) { gpr_mu_unlock(&sp->server->mu); deactivated_all_ports(exec_ctx, sp->server); @@ -300,34 +300,37 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { /* Re-arm the notification event so we get another chance to read. */ grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure); + gpr_mu_unlock(&sp->server->mu); } static int add_socket_to_server(grpc_udp_server *s, int fd, - const struct sockaddr *addr, size_t addr_len, + const grpc_resolved_address *addr, grpc_udp_server_read_cb read_cb, grpc_udp_server_orphan_cb orphan_cb) { - server_port *sp; + grpc_udp_listener *sp; int port; char *addr_str; char *name; - port = prepare_socket(fd, addr, addr_len); + port = prepare_socket(fd, addr); if (port >= 0) { - grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1); + grpc_sockaddr_to_string(&addr_str, addr, 1); gpr_asprintf(&name, "udp-server-listener:%s", addr_str); gpr_free(addr_str); gpr_mu_lock(&s->mu); - /* append it to the list under a lock */ - if (s->nports == s->port_capacity) { - s->port_capacity *= 2; - s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity); + s->nports++; + sp = gpr_malloc(sizeof(grpc_udp_listener)); + sp->next = NULL; + if (s->head == NULL) { + s->head = sp; + } else { + s->tail->next = sp; } - sp = &s->ports[s->nports++]; + s->tail = sp; sp->server = s; sp->fd = fd; sp->emfd = grpc_fd_create(fd, name); - memcpy(sp->addr.untyped, addr, addr_len); - sp->addr_len = addr_len; + memcpy(&sp->addr, addr, sizeof(grpc_resolved_address)); sp->read_cb = read_cb; sp->orphan_cb = orphan_cb; GPR_ASSERT(sp->emfd); @@ -338,34 +341,34 @@ static int add_socket_to_server(grpc_udp_server *s, int fd, return port; } -int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, - size_t addr_len, grpc_udp_server_read_cb read_cb, +int grpc_udp_server_add_port(grpc_udp_server *s, + const grpc_resolved_address *addr, + grpc_udp_server_read_cb read_cb, grpc_udp_server_orphan_cb orphan_cb) { + grpc_udp_listener *sp; int allocated_port1 = -1; int allocated_port2 = -1; - unsigned i; int fd; grpc_dualstack_mode dsmode; - struct sockaddr_in6 addr6_v4mapped; - struct sockaddr_in wild4; - struct sockaddr_in6 wild6; - struct sockaddr_in addr4_copy; - struct sockaddr *allocated_addr = NULL; - struct sockaddr_storage sockname_temp; - socklen_t sockname_len; + grpc_resolved_address addr6_v4mapped; + grpc_resolved_address wild4; + grpc_resolved_address wild6; + grpc_resolved_address addr4_copy; + grpc_resolved_address *allocated_addr = NULL; + grpc_resolved_address sockname_temp; int port; /* 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) { - for (i = 0; i < s->nports; i++) { - sockname_len = sizeof(sockname_temp); - if (0 == getsockname(s->ports[i].fd, (struct sockaddr *)&sockname_temp, - &sockname_len)) { - port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + for (sp = s->head; sp; sp = sp->next) { + sockname_temp.len = sizeof(struct sockaddr_storage); + if (0 == getsockname(sp->fd, (struct sockaddr *)sockname_temp.addr, + (socklen_t *)&sockname_temp.len)) { + port = grpc_sockaddr_get_port(&sockname_temp); if (port > 0) { - allocated_addr = gpr_malloc(addr_len); - memcpy(allocated_addr, addr, addr_len); + allocated_addr = 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; @@ -375,8 +378,7 @@ int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, } if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { - addr = (const struct sockaddr *)&addr6_v4mapped; - addr_len = sizeof(addr6_v4mapped); + addr = &addr6_v4mapped; } /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */ @@ -384,22 +386,19 @@ int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, grpc_sockaddr_make_wildcards(port, &wild4, &wild6); /* Try listening on IPv6 first. */ - addr = (struct sockaddr *)&wild6; - addr_len = sizeof(wild6); + addr = &wild6; // TODO(rjshade): Test and propagate the returned grpc_error*: grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd); - allocated_port1 = - add_socket_to_server(s, fd, addr, addr_len, read_cb, orphan_cb); + allocated_port1 = add_socket_to_server(s, fd, addr, read_cb, orphan_cb); if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) { goto done; } /* If we didn't get a dualstack socket, also listen on 0.0.0.0. */ if (port == 0 && allocated_port1 > 0) { - grpc_sockaddr_set_port((struct sockaddr *)&wild4, allocated_port1); + grpc_sockaddr_set_port(&wild4, allocated_port1); } - addr = (struct sockaddr *)&wild4; - addr_len = sizeof(wild4); + addr = &wild4; } // TODO(rjshade): Test and propagate the returned grpc_error*: @@ -409,11 +408,9 @@ int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, } if (dsmode == GRPC_DSMODE_IPV4 && grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { - addr = (struct sockaddr *)&addr4_copy; - addr_len = sizeof(addr4_copy); + addr = &addr4_copy; } - allocated_port2 = - add_socket_to_server(s, fd, addr, addr_len, read_cb, orphan_cb); + allocated_port2 = add_socket_to_server(s, fd, addr, read_cb, orphan_cb); done: gpr_free(allocated_addr); @@ -421,27 +418,40 @@ done: } int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) { - return (port_index < s->nports) ? s->ports[port_index].fd : -1; + grpc_udp_listener *sp; + if (port_index >= s->nports) { + return -1; + } + + for (sp = s->head; sp && port_index != 0; sp = sp->next) { + --port_index; + } + return sp->fd; } void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, grpc_pollset **pollsets, size_t pollset_count, grpc_server *server) { - size_t i, j; + size_t i; gpr_mu_lock(&s->mu); + grpc_udp_listener *sp; GPR_ASSERT(s->active_ports == 0); s->pollsets = pollsets; s->grpc_server = server; - for (i = 0; i < s->nports; i++) { - for (j = 0; j < pollset_count; j++) { - grpc_pollset_add_fd(exec_ctx, pollsets[j], s->ports[i].emfd); + + sp = s->head; + while (sp != NULL) { + for (i = 0; i < pollset_count; i++) { + grpc_pollset_add_fd(exec_ctx, pollsets[i], sp->emfd); } - s->ports[i].read_closure.cb = on_read; - s->ports[i].read_closure.cb_arg = &s->ports[i]; - grpc_fd_notify_on_read(exec_ctx, s->ports[i].emfd, - &s->ports[i].read_closure); + sp->read_closure.cb = on_read; + sp->read_closure.cb_arg = sp; + grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure); + s->active_ports++; + sp = sp->next; } + gpr_mu_unlock(&s->mu); } |