LCOV - code coverage report
Current view: top level - core/iomgr - udp_server.c (source / functions) Hit Total Coverage
Test: tmp.CaZ6RjdVn2 Lines: 131 179 73.2 %
Date: 2015-12-10 22:15:08 Functions: 11 13 84.6 %

          Line data    Source code
       1             : /*
       2             :  *
       3             :  * Copyright 2015, Google Inc.
       4             :  * All rights reserved.
       5             :  *
       6             :  * Redistribution and use in source and binary forms, with or without
       7             :  * modification, are permitted provided that the following conditions are
       8             :  * met:
       9             :  *
      10             :  *     * Redistributions of source code must retain the above copyright
      11             :  * notice, this list of conditions and the following disclaimer.
      12             :  *     * Redistributions in binary form must reproduce the above
      13             :  * copyright notice, this list of conditions and the following disclaimer
      14             :  * in the documentation and/or other materials provided with the
      15             :  * distribution.
      16             :  *     * Neither the name of Google Inc. nor the names of its
      17             :  * contributors may be used to endorse or promote products derived from
      18             :  * this software without specific prior written permission.
      19             :  *
      20             :  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
      21             :  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
      22             :  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
      23             :  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
      24             :  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
      25             :  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
      26             :  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
      27             :  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
      28             :  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
      29             :  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
      30             :  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
      31             :  *
      32             :  */
      33             : 
      34             : /* FIXME: "posix" files shouldn't be depending on _GNU_SOURCE */
      35             : #ifndef _GNU_SOURCE
      36             : #define _GNU_SOURCE
      37             : #endif
      38             : 
      39             : #include <grpc/support/port_platform.h>
      40             : 
      41             : #ifdef GPR_POSIX_SOCKET
      42             : 
      43             : #include "src/core/iomgr/udp_server.h"
      44             : 
      45             : #include <errno.h>
      46             : #include <fcntl.h>
      47             : #include <limits.h>
      48             : #include <netinet/in.h>
      49             : #include <netinet/tcp.h>
      50             : #include <string.h>
      51             : #include <sys/socket.h>
      52             : #include <sys/stat.h>
      53             : #include <sys/types.h>
      54             : #include <sys/un.h>
      55             : #include <unistd.h>
      56             : 
      57             : #include "src/core/iomgr/fd_posix.h"
      58             : #include "src/core/iomgr/pollset_posix.h"
      59             : #include "src/core/iomgr/resolve_address.h"
      60             : #include "src/core/iomgr/sockaddr_utils.h"
      61             : #include "src/core/iomgr/socket_utils_posix.h"
      62             : #include "src/core/support/string.h"
      63             : #include <grpc/support/alloc.h>
      64             : #include <grpc/support/log.h>
      65             : #include <grpc/support/sync.h>
      66             : #include <grpc/support/string_util.h>
      67             : #include <grpc/support/time.h>
      68             : 
      69             : #define INIT_PORT_CAP 2
      70             : 
      71             : /* one listening port */
      72             : typedef struct {
      73             :   int fd;
      74             :   grpc_fd *emfd;
      75             :   grpc_udp_server *server;
      76             :   union {
      77             :     gpr_uint8 untyped[GRPC_MAX_SOCKADDR_SIZE];
      78             :     struct sockaddr sockaddr;
      79             :     struct sockaddr_un un;
      80             :   } addr;
      81             :   size_t addr_len;
      82             :   grpc_closure read_closure;
      83             :   grpc_closure destroyed_closure;
      84             :   grpc_udp_server_read_cb read_cb;
      85             : } server_port;
      86             : 
      87           0 : static void unlink_if_unix_domain_socket(const struct sockaddr_un *un) {
      88             :   struct stat st;
      89             : 
      90           0 :   if (stat(un->sun_path, &st) == 0 && (st.st_mode & S_IFMT) == S_IFSOCK) {
      91           0 :     unlink(un->sun_path);
      92             :   }
      93           0 : }
      94             : 
      95             : /* the overall server */
      96             : struct grpc_udp_server {
      97             :   gpr_mu mu;
      98             :   gpr_cv cv;
      99             : 
     100             :   /* active port count: how many ports are actually still listening */
     101             :   size_t active_ports;
     102             :   /* destroyed port count: how many ports are completely destroyed */
     103             :   size_t destroyed_ports;
     104             : 
     105             :   /* is this server shutting down? (boolean) */
     106             :   int shutdown;
     107             : 
     108             :   /* all listening ports */
     109             :   server_port *ports;
     110             :   size_t nports;
     111             :   size_t port_capacity;
     112             : 
     113             :   /* shutdown callback */
     114             :   grpc_closure *shutdown_complete;
     115             : 
     116             :   /* all pollsets interested in new connections */
     117             :   grpc_pollset **pollsets;
     118             :   /* number of pollsets in the pollsets array */
     119             :   size_t pollset_count;
     120             :   /* The parent grpc server */
     121             :   grpc_server *grpc_server;
     122             : };
     123             : 
     124           6 : grpc_udp_server *grpc_udp_server_create(void) {
     125           6 :   grpc_udp_server *s = gpr_malloc(sizeof(grpc_udp_server));
     126           6 :   gpr_mu_init(&s->mu);
     127           6 :   gpr_cv_init(&s->cv);
     128           6 :   s->active_ports = 0;
     129           6 :   s->destroyed_ports = 0;
     130           6 :   s->shutdown = 0;
     131           6 :   s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP);
     132           6 :   s->nports = 0;
     133           6 :   s->port_capacity = INIT_PORT_CAP;
     134             : 
     135           6 :   return s;
     136             : }
     137             : 
     138           6 : static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
     139           6 :   grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
     140             : 
     141           6 :   gpr_mu_destroy(&s->mu);
     142           6 :   gpr_cv_destroy(&s->cv);
     143             : 
     144           6 :   gpr_free(s->ports);
     145           6 :   gpr_free(s);
     146           6 : }
     147             : 
     148           4 : static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server, int success) {
     149           4 :   grpc_udp_server *s = server;
     150           4 :   gpr_mu_lock(&s->mu);
     151           4 :   s->destroyed_ports++;
     152           4 :   if (s->destroyed_ports == s->nports) {
     153           4 :     gpr_mu_unlock(&s->mu);
     154           4 :     finish_shutdown(exec_ctx, s);
     155             :   } else {
     156           0 :     gpr_mu_unlock(&s->mu);
     157             :   }
     158           4 : }
     159             : 
     160             : /* called when all listening endpoints have been shutdown, so no further
     161             :    events will be received on them - at this point it's safe to destroy
     162             :    things */
     163           6 : static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
     164             :   size_t i;
     165             : 
     166             :   /* delete ALL the things */
     167           6 :   gpr_mu_lock(&s->mu);
     168             : 
     169           6 :   if (!s->shutdown) {
     170           0 :     gpr_mu_unlock(&s->mu);
     171           6 :     return;
     172             :   }
     173             : 
     174           6 :   if (s->nports) {
     175           8 :     for (i = 0; i < s->nports; i++) {
     176           4 :       server_port *sp = &s->ports[i];
     177           4 :       if (sp->addr.sockaddr.sa_family == AF_UNIX) {
     178           0 :         unlink_if_unix_domain_socket(&sp->addr.un);
     179             :       }
     180           4 :       sp->destroyed_closure.cb = destroyed_port;
     181           4 :       sp->destroyed_closure.cb_arg = s;
     182           4 :       grpc_fd_orphan(exec_ctx, sp->emfd, &sp->destroyed_closure, NULL,
     183             :                      "udp_listener_shutdown");
     184             :     }
     185           4 :     gpr_mu_unlock(&s->mu);
     186             :   } else {
     187           2 :     gpr_mu_unlock(&s->mu);
     188           2 :     finish_shutdown(exec_ctx, s);
     189             :   }
     190             : }
     191             : 
     192           6 : void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
     193             :                              grpc_closure *on_done) {
     194             :   size_t i;
     195           6 :   gpr_mu_lock(&s->mu);
     196             : 
     197           6 :   GPR_ASSERT(!s->shutdown);
     198           6 :   s->shutdown = 1;
     199             : 
     200           6 :   s->shutdown_complete = on_done;
     201             : 
     202             :   /* shutdown all fd's */
     203           6 :   if (s->active_ports) {
     204           6 :     for (i = 0; i < s->nports; i++) {
     205           3 :       grpc_fd_shutdown(exec_ctx, s->ports[i].emfd);
     206             :     }
     207           3 :     gpr_mu_unlock(&s->mu);
     208             :   } else {
     209           3 :     gpr_mu_unlock(&s->mu);
     210           3 :     deactivated_all_ports(exec_ctx, s);
     211             :   }
     212           6 : }
     213             : 
     214             : /* Prepare a recently-created socket for listening. */
     215           4 : static int prepare_socket(int fd, const struct sockaddr *addr,
     216             :                           size_t addr_len) {
     217             :   struct sockaddr_storage sockname_temp;
     218             :   socklen_t sockname_len;
     219             :   int get_local_ip;
     220             :   int rc;
     221             : 
     222           4 :   if (fd < 0) {
     223           0 :     goto error;
     224             :   }
     225             : 
     226           4 :   if (!grpc_set_socket_nonblocking(fd, 1) || !grpc_set_socket_cloexec(fd, 1)) {
     227           0 :     gpr_log(GPR_ERROR, "Unable to configure socket %d: %s", fd,
     228           0 :             strerror(errno));
     229             :   }
     230             : 
     231           4 :   get_local_ip = 1;
     232           4 :   rc = setsockopt(fd, IPPROTO_IP, IP_PKTINFO, &get_local_ip,
     233             :                   sizeof(get_local_ip));
     234           4 :   if (rc == 0 && addr->sa_family == AF_INET6) {
     235             : #if !defined(__APPLE__)
     236           4 :     rc = setsockopt(fd, IPPROTO_IPV6, IPV6_RECVPKTINFO, &get_local_ip,
     237             :                     sizeof(get_local_ip));
     238             : #endif
     239             :   }
     240             : 
     241           4 :   GPR_ASSERT(addr_len < ~(socklen_t)0);
     242           4 :   if (bind(fd, addr, (socklen_t)addr_len) < 0) {
     243             :     char *addr_str;
     244           0 :     grpc_sockaddr_to_string(&addr_str, addr, 0);
     245           0 :     gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno));
     246           0 :     gpr_free(addr_str);
     247           0 :     goto error;
     248             :   }
     249             : 
     250           4 :   sockname_len = sizeof(sockname_temp);
     251           4 :   if (getsockname(fd, (struct sockaddr *)&sockname_temp, &sockname_len) < 0) {
     252           0 :     goto error;
     253             :   }
     254             : 
     255           4 :   return grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp);
     256             : 
     257             : error:
     258           0 :   if (fd >= 0) {
     259           0 :     close(fd);
     260             :   }
     261           0 :   return -1;
     262             : }
     263             : 
     264             : /* event manager callback when reads are ready */
     265          14 : static void on_read(grpc_exec_ctx *exec_ctx, void *arg, int success) {
     266          14 :   server_port *sp = arg;
     267             : 
     268          14 :   if (success == 0) {
     269           3 :     gpr_mu_lock(&sp->server->mu);
     270           3 :     if (0 == --sp->server->active_ports) {
     271           3 :       gpr_mu_unlock(&sp->server->mu);
     272           3 :       deactivated_all_ports(exec_ctx, sp->server);
     273             :     } else {
     274           0 :       gpr_mu_unlock(&sp->server->mu);
     275             :     }
     276          17 :     return;
     277             :   }
     278             : 
     279             :   /* Tell the registered callback that data is available to read. */
     280          11 :   GPR_ASSERT(sp->read_cb);
     281          11 :   sp->read_cb(exec_ctx, sp->emfd, sp->server->grpc_server);
     282             : 
     283             :   /* Re-arm the notification event so we get another chance to read. */
     284          11 :   grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
     285             : }
     286             : 
     287           4 : static int add_socket_to_server(grpc_udp_server *s, int fd,
     288             :                                 const struct sockaddr *addr, size_t addr_len,
     289             :                                 grpc_udp_server_read_cb read_cb) {
     290             :   server_port *sp;
     291             :   int port;
     292             :   char *addr_str;
     293             :   char *name;
     294             : 
     295           4 :   port = prepare_socket(fd, addr, addr_len);
     296           4 :   if (port >= 0) {
     297           4 :     grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1);
     298           4 :     gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
     299           4 :     gpr_free(addr_str);
     300           4 :     gpr_mu_lock(&s->mu);
     301             :     /* append it to the list under a lock */
     302           4 :     if (s->nports == s->port_capacity) {
     303           0 :       s->port_capacity *= 2;
     304           0 :       s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity);
     305             :     }
     306           4 :     sp = &s->ports[s->nports++];
     307           4 :     sp->server = s;
     308           4 :     sp->fd = fd;
     309           4 :     sp->emfd = grpc_fd_create(fd, name);
     310           4 :     memcpy(sp->addr.untyped, addr, addr_len);
     311           4 :     sp->addr_len = addr_len;
     312           4 :     sp->read_cb = read_cb;
     313           4 :     GPR_ASSERT(sp->emfd);
     314           4 :     gpr_mu_unlock(&s->mu);
     315           4 :     gpr_free(name);
     316             :   }
     317             : 
     318           4 :   return port;
     319             : }
     320             : 
     321           4 : int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr,
     322             :                              size_t addr_len, grpc_udp_server_read_cb read_cb) {
     323           4 :   int allocated_port1 = -1;
     324           4 :   int allocated_port2 = -1;
     325             :   unsigned i;
     326             :   int fd;
     327             :   grpc_dualstack_mode dsmode;
     328             :   struct sockaddr_in6 addr6_v4mapped;
     329             :   struct sockaddr_in wild4;
     330             :   struct sockaddr_in6 wild6;
     331             :   struct sockaddr_in addr4_copy;
     332           4 :   struct sockaddr *allocated_addr = NULL;
     333             :   struct sockaddr_storage sockname_temp;
     334             :   socklen_t sockname_len;
     335             :   int port;
     336             : 
     337           4 :   if (((struct sockaddr *)addr)->sa_family == AF_UNIX) {
     338           0 :     unlink_if_unix_domain_socket(addr);
     339             :   }
     340             : 
     341             :   /* Check if this is a wildcard port, and if so, try to keep the port the same
     342             :      as some previously created listener. */
     343           4 :   if (grpc_sockaddr_get_port(addr) == 0) {
     344           4 :     for (i = 0; i < s->nports; i++) {
     345           0 :       sockname_len = sizeof(sockname_temp);
     346           0 :       if (0 == getsockname(s->ports[i].fd, (struct sockaddr *)&sockname_temp,
     347             :                            &sockname_len)) {
     348           0 :         port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp);
     349           0 :         if (port > 0) {
     350           0 :           allocated_addr = malloc(addr_len);
     351           0 :           memcpy(allocated_addr, addr, addr_len);
     352           0 :           grpc_sockaddr_set_port(allocated_addr, port);
     353           0 :           addr = allocated_addr;
     354           0 :           break;
     355             :         }
     356             :       }
     357             :     }
     358             :   }
     359             : 
     360           4 :   if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
     361           4 :     addr = (const struct sockaddr *)&addr6_v4mapped;
     362           4 :     addr_len = sizeof(addr6_v4mapped);
     363             :   }
     364             : 
     365             :   /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
     366           4 :   if (grpc_sockaddr_is_wildcard(addr, &port)) {
     367           4 :     grpc_sockaddr_make_wildcards(port, &wild4, &wild6);
     368             : 
     369             :     /* Try listening on IPv6 first. */
     370           4 :     addr = (struct sockaddr *)&wild6;
     371           4 :     addr_len = sizeof(wild6);
     372           4 :     fd = grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode);
     373           4 :     allocated_port1 = add_socket_to_server(s, fd, addr, addr_len, read_cb);
     374           4 :     if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) {
     375           4 :       goto done;
     376             :     }
     377             : 
     378             :     /* If we didn't get a dualstack socket, also listen on 0.0.0.0. */
     379           0 :     if (port == 0 && allocated_port1 > 0) {
     380           0 :       grpc_sockaddr_set_port((struct sockaddr *)&wild4, allocated_port1);
     381             :     }
     382           0 :     addr = (struct sockaddr *)&wild4;
     383           0 :     addr_len = sizeof(wild4);
     384             :   }
     385             : 
     386           0 :   fd = grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode);
     387           0 :   if (fd < 0) {
     388           0 :     gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
     389             :   }
     390           0 :   if (dsmode == GRPC_DSMODE_IPV4 &&
     391           0 :       grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) {
     392           0 :     addr = (struct sockaddr *)&addr4_copy;
     393           0 :     addr_len = sizeof(addr4_copy);
     394             :   }
     395           0 :   allocated_port2 = add_socket_to_server(s, fd, addr, addr_len, read_cb);
     396             : 
     397             : done:
     398           4 :   gpr_free(allocated_addr);
     399           4 :   return allocated_port1 >= 0 ? allocated_port1 : allocated_port2;
     400             : }
     401             : 
     402           2 : int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) {
     403           2 :   return (port_index < s->nports) ? s->ports[port_index].fd : -1;
     404             : }
     405             : 
     406           4 : void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
     407             :                            grpc_pollset **pollsets, size_t pollset_count,
     408             :                            grpc_server *server) {
     409             :   size_t i, j;
     410           4 :   gpr_mu_lock(&s->mu);
     411           4 :   GPR_ASSERT(s->active_ports == 0);
     412           4 :   s->pollsets = pollsets;
     413           4 :   s->grpc_server = server;
     414           7 :   for (i = 0; i < s->nports; i++) {
     415           5 :     for (j = 0; j < pollset_count; j++) {
     416           2 :       grpc_pollset_add_fd(exec_ctx, pollsets[j], s->ports[i].emfd);
     417             :     }
     418           3 :     s->ports[i].read_closure.cb = on_read;
     419           3 :     s->ports[i].read_closure.cb_arg = &s->ports[i];
     420           3 :     grpc_fd_notify_on_read(exec_ctx, s->ports[i].emfd,
     421           3 :                            &s->ports[i].read_closure);
     422           3 :     s->active_ports++;
     423             :   }
     424           4 :   gpr_mu_unlock(&s->mu);
     425           4 : }
     426             : 
     427             : /* TODO(rjshade): Add a test for this method. */
     428           0 : void grpc_udp_server_write(server_port *sp, const char *buffer, size_t buf_len,
     429             :                            const struct sockaddr *peer_address) {
     430             :   ssize_t rc;
     431           0 :   rc = sendto(sp->fd, buffer, buf_len, 0, peer_address, sizeof(peer_address));
     432           0 :   if (rc < 0) {
     433           0 :     gpr_log(GPR_ERROR, "Unable to send data: %s", strerror(errno));
     434             :   }
     435           0 : }
     436             : 
     437             : #endif

Generated by: LCOV version 1.11