diff options
| -rw-r--r-- | include/grpc/impl/codegen/grpc_types.h | 2 | ||||
| -rw-r--r-- | src/core/lib/iomgr/tcp_server_posix.c | 331 | ||||
| -rw-r--r-- | test/core/handshake/client_ssl.c | 27 | ||||
| -rw-r--r-- | test/core/iomgr/tcp_server_posix_test.c | 294 |
4 files changed, 489 insertions, 165 deletions
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index bc0b2bd7f8..6661547f6b 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -222,6 +222,8 @@ typedef struct { /** If non-zero, a pointer to a buffer pool (use grpc_resource_quota_arg_vtable to fetch an appropriate pointer arg vtable) */ #define GRPC_ARG_RESOURCE_QUOTA "grpc.resource_quota" +/** If non-zero, expand wildcard addresses to a list of local addresses. */ +#define GRPC_ARG_EXPAND_WILDCARD_ADDRS "grpc.expand_wildcard_addrs" /** Service config data in JSON form. Not intended for use outside of tests. */ #define GRPC_ARG_SERVICE_CONFIG "grpc.service_config" /** LB policy name. */ diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index e9e7511c9c..36f878fdd4 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -44,6 +44,7 @@ #include <errno.h> #include <fcntl.h> +#include <ifaddrs.h> #include <limits.h> #include <netinet/in.h> #include <netinet/tcp.h> @@ -115,6 +116,8 @@ struct grpc_tcp_server { bool shutdown; /* use SO_REUSEPORT */ bool so_reuseport; + /* expand wildcard addresses to a list of all local addresses */ + bool expand_wildcard_addrs; /* linked list of server ports */ grpc_tcp_listener *head; @@ -161,6 +164,7 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); s->so_reuseport = has_so_reuseport; s->resource_quota = grpc_resource_quota_create(NULL); + s->expand_wildcard_addrs = false; for (size_t i = 0; i < (args == NULL ? 0 : args->num_args); i++) { if (0 == strcmp(GRPC_ARG_ALLOW_REUSEPORT, args->args[i].key)) { if (args->args[i].type == GRPC_ARG_INTEGER) { @@ -183,6 +187,15 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA " must be a pointer to a buffer pool"); } + } else if (0 == strcmp(GRPC_ARG_EXPAND_WILDCARD_ADDRS, args->args[i].key)) { + if (args->args[i].type == GRPC_ARG_INTEGER) { + s->expand_wildcard_addrs = (args->args[i].value.integer != 0); + } else { + grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota); + gpr_free(s); + return GRPC_ERROR_CREATE(GRPC_ARG_EXPAND_WILDCARD_ADDRS + " must be an integer"); + } } } gpr_ref_init(&s->refs, 1); @@ -504,9 +517,224 @@ static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd, return err; } -/* Insert count new listeners after listener. Every new listener will have the - same listen address as listener (SO_REUSEPORT must be enabled). Every new - listener is a sibling of listener. */ +/* If successful, add a listener to s for addr, set *dsmode for the socket, and + return the *listener. */ +static grpc_error *add_addr_to_server(grpc_tcp_server *s, + const grpc_resolved_address *addr, + unsigned port_index, unsigned fd_index, + grpc_dualstack_mode *dsmode, + grpc_tcp_listener **listener) { + grpc_resolved_address addr4_copy; + int fd; + grpc_error *err = + grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, dsmode, &fd); + if (err != GRPC_ERROR_NONE) { + return err; + } + if (*dsmode == GRPC_DSMODE_IPV4 && + grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { + addr = &addr4_copy; + } + return add_socket_to_server(s, fd, addr, port_index, fd_index, listener); +} + +/* Bind to "::" to get a port number not used by any address. */ +static grpc_error *get_unused_port(int *port) { + grpc_resolved_address wild; + grpc_sockaddr_make_wildcard6(0, &wild); + grpc_dualstack_mode dsmode; + int fd; + grpc_error *err = + grpc_create_dualstack_socket(&wild, SOCK_STREAM, 0, &dsmode, &fd); + if (err != GRPC_ERROR_NONE) { + return err; + } + if (dsmode == GRPC_DSMODE_IPV4) { + grpc_sockaddr_make_wildcard4(0, &wild); + } + if (bind(fd, (const struct sockaddr *)wild.addr, (socklen_t)wild.len) != 0) { + err = GRPC_OS_ERROR(errno, "bind"); + close(fd); + return err; + } + if (getsockname(fd, (struct sockaddr *)wild.addr, (socklen_t *)&wild.len) != + 0) { + err = GRPC_OS_ERROR(errno, "getsockname"); + close(fd); + return err; + } + close(fd); + *port = grpc_sockaddr_get_port(&wild); + return *port <= 0 ? GRPC_ERROR_CREATE("Bad port") : GRPC_ERROR_NONE; +} + +/* Return the listener in s with address addr or NULL. */ +static grpc_tcp_listener *find_listener_with_addr(grpc_tcp_server *s, + grpc_resolved_address *addr) { + grpc_tcp_listener *l; + gpr_mu_lock(&s->mu); + for (l = s->head; l != NULL; l = l->next) { + if (l->addr.len != addr->len) { + continue; + } + if (memcmp(l->addr.addr, addr->addr, addr->len) == 0) { + break; + } + } + gpr_mu_unlock(&s->mu); + return l; +} + +/* Get all addresses assigned to network interfaces on the machine and create a + listener for each. requested_port is the port to use for every listener, or 0 + to select one random port that will be used for every listener. Set *out_port + to the port selected. Return GRPC_ERROR_NONE only if all listeners were + added. */ +static grpc_error *add_all_local_addrs_to_server(grpc_tcp_server *s, + unsigned port_index, + int requested_port, + int *out_port) { + struct ifaddrs *ifa = NULL; + struct ifaddrs *ifa_it; + unsigned fd_index = 0; + grpc_tcp_listener *sp = NULL; + grpc_error *err = GRPC_ERROR_NONE; + if (requested_port == 0) { + /* Note: There could be a race where some local addrs can listen on the + selected port and some can't. The sane way to handle this would be to + retry by recreating the whole grpc_tcp_server. Backing out individual + listeners and orphaning the FDs looks like too much trouble. */ + if ((err = get_unused_port(&requested_port)) != GRPC_ERROR_NONE) { + return err; + } else if (requested_port <= 0) { + return GRPC_ERROR_CREATE("Bad get_unused_port()"); + } + gpr_log(GPR_DEBUG, "Picked unused port %d", requested_port); + } + if (getifaddrs(&ifa) != 0 || ifa == NULL) { + return GRPC_OS_ERROR(errno, "getifaddrs"); + } + for (ifa_it = ifa; ifa_it != NULL; ifa_it = ifa_it->ifa_next) { + grpc_resolved_address addr; + char *addr_str = NULL; + grpc_dualstack_mode dsmode; + grpc_tcp_listener *new_sp = NULL; + const char *ifa_name = (ifa_it->ifa_name ? ifa_it->ifa_name : "<unknown>"); + if (ifa_it->ifa_addr == NULL) { + continue; + } else if (ifa_it->ifa_addr->sa_family == AF_INET) { + addr.len = sizeof(struct sockaddr_in); + } else if (ifa_it->ifa_addr->sa_family == AF_INET6) { + addr.len = sizeof(struct sockaddr_in6); + } else { + continue; + } + memcpy(addr.addr, ifa_it->ifa_addr, addr.len); + if (!grpc_sockaddr_set_port(&addr, requested_port)) { + /* Should never happen, because we check sa_family above. */ + err = GRPC_ERROR_CREATE("Failed to set port"); + break; + } + if (grpc_sockaddr_to_string(&addr_str, &addr, 0) < 0) { + addr_str = gpr_strdup("<error>"); + } + gpr_log(GPR_DEBUG, + "Adding local addr from interface %s flags 0x%x to server: %s", + ifa_name, ifa_it->ifa_flags, addr_str); + /* We could have multiple interfaces with the same address (e.g., bonding), + so look for duplicates. */ + if (find_listener_with_addr(s, &addr) != NULL) { + gpr_log(GPR_DEBUG, "Skipping duplicate addr %s on interface %s", addr_str, + ifa_name); + gpr_free(addr_str); + continue; + } + if ((err = add_addr_to_server(s, &addr, port_index, fd_index, &dsmode, + &new_sp)) != GRPC_ERROR_NONE) { + char *err_str = NULL; + grpc_error *root_err; + if (gpr_asprintf(&err_str, "Failed to add listener: %s", addr_str) < 0) { + err_str = gpr_strdup("Failed to add listener"); + } + root_err = GRPC_ERROR_CREATE(err_str); + gpr_free(err_str); + gpr_free(addr_str); + err = grpc_error_add_child(root_err, err); + break; + } else { + GPR_ASSERT(requested_port == new_sp->port); + ++fd_index; + if (sp != NULL) { + new_sp->is_sibling = 1; + sp->sibling = new_sp; + } + sp = new_sp; + } + gpr_free(addr_str); + } + freeifaddrs(ifa); + if (err != GRPC_ERROR_NONE) { + return err; + } else if (sp == NULL) { + return GRPC_ERROR_CREATE("No local addresses"); + } else { + *out_port = sp->port; + return GRPC_ERROR_NONE; + } +} + +/* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */ +static grpc_error *add_wildcard_addrs_to_server(grpc_tcp_server *s, + unsigned port_index, + int requested_port, + int *out_port) { + grpc_resolved_address wild4; + grpc_resolved_address wild6; + unsigned fd_index = 0; + grpc_dualstack_mode dsmode; + grpc_tcp_listener *sp = NULL; + grpc_tcp_listener *sp2 = NULL; + grpc_error *v6_err = GRPC_ERROR_NONE; + grpc_error *v4_err = GRPC_ERROR_NONE; + *out_port = -1; + if (s->expand_wildcard_addrs) { + return add_all_local_addrs_to_server(s, port_index, requested_port, + out_port); + } + grpc_sockaddr_make_wildcards(requested_port, &wild4, &wild6); + /* Try listening on IPv6 first. */ + if ((v6_err = add_addr_to_server(s, &wild6, port_index, fd_index, &dsmode, + &sp)) == GRPC_ERROR_NONE) { + ++fd_index; + requested_port = *out_port = sp->port; + if (dsmode == GRPC_DSMODE_DUALSTACK || dsmode == GRPC_DSMODE_IPV4) { + return GRPC_ERROR_NONE; + } + } + /* If we got a v6-only socket or nothing, try adding 0.0.0.0. */ + grpc_sockaddr_set_port(&wild4, requested_port); + if ((v4_err = add_addr_to_server(s, &wild4, port_index, fd_index, &dsmode, + &sp2)) == GRPC_ERROR_NONE) { + *out_port = sp2->port; + if (sp != NULL) { + sp2->is_sibling = 1; + sp->sibling = sp2; + } + } + if (*out_port > 0) { + GRPC_LOG_IF_ERROR("Failed to add :: listener", v6_err); + GRPC_LOG_IF_ERROR("Failed to add 0.0.0.0 listener", v4_err); + return GRPC_ERROR_NONE; + } else { + grpc_error *root_err = + GRPC_ERROR_CREATE("Failed to add any wildcard listeners"); + GPR_ASSERT(v6_err != GRPC_ERROR_NONE && v4_err != GRPC_ERROR_NONE); + root_err = grpc_error_add_child(root_err, v6_err); + root_err = grpc_error_add_child(root_err, v4_err); + return root_err; + } +} + static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) { grpc_tcp_listener *sp = NULL; char *addr_str; @@ -559,19 +787,13 @@ grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s, const grpc_resolved_address *addr, int *out_port) { grpc_tcp_listener *sp; - grpc_tcp_listener *sp2 = NULL; - 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 = NULL; grpc_resolved_address sockname_temp; - int port; + grpc_resolved_address addr6_v4mapped; + int requested_port = grpc_sockaddr_get_port(addr); unsigned port_index = 0; - unsigned fd_index = 0; - grpc_error *errs[2] = {GRPC_ERROR_NONE, GRPC_ERROR_NONE}; + grpc_dualstack_mode dsmode; + grpc_error *err; + *out_port = -1; if (s->tail != NULL) { port_index = s->tail->port_index + 1; } @@ -579,85 +801,34 @@ grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s, /* 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) { + if (requested_port == 0) { 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, + 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(sizeof(grpc_resolved_address)); - memcpy(allocated_addr, addr, addr->len); - grpc_sockaddr_set_port(allocated_addr, port); - addr = allocated_addr; + int used_port = grpc_sockaddr_get_port(&sockname_temp); + if (used_port > 0) { + memcpy(&sockname_temp, addr, sizeof(grpc_resolved_address)); + grpc_sockaddr_set_port(&sockname_temp, used_port); + requested_port = used_port; + addr = &sockname_temp; break; } } } } - - sp = NULL; - + if (grpc_sockaddr_is_wildcard(addr, &requested_port)) { + return add_wildcard_addrs_to_server(s, port_index, requested_port, + out_port); + } if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { addr = &addr6_v4mapped; } - - /* 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; - errs[0] = grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, &dsmode, &fd); - if (errs[0] == GRPC_ERROR_NONE) { - errs[0] = add_socket_to_server(s, fd, addr, port_index, fd_index, &sp); - if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) { - goto done; - } - if (sp != NULL) { - ++fd_index; - } - /* If we didn't get a dualstack socket, also listen on 0.0.0.0. */ - if (port == 0 && sp != NULL) { - grpc_sockaddr_set_port(&wild4, sp->port); - } - } - addr = &wild4; - } - - errs[1] = grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, &dsmode, &fd); - if (errs[1] == GRPC_ERROR_NONE) { - if (dsmode == GRPC_DSMODE_IPV4 && - grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { - addr = &addr4_copy; - } - sp2 = sp; - errs[1] = add_socket_to_server(s, fd, addr, port_index, fd_index, &sp); - if (sp2 != NULL && sp != NULL) { - sp2->sibling = sp; - sp->is_sibling = 1; - } - } - -done: - gpr_free(allocated_addr); - if (sp != NULL) { + if ((err = add_addr_to_server(s, addr, port_index, 0, &dsmode, &sp)) == + GRPC_ERROR_NONE) { *out_port = sp->port; - GRPC_ERROR_UNREF(errs[0]); - GRPC_ERROR_UNREF(errs[1]); - return GRPC_ERROR_NONE; - } else { - *out_port = -1; - char *addr_str = grpc_sockaddr_to_uri(addr); - grpc_error *err = grpc_error_set_str( - GRPC_ERROR_CREATE_REFERENCING("Failed to add port to server", errs, - GPR_ARRAY_SIZE(errs)), - GRPC_ERROR_STR_TARGET_ADDRESS, addr_str); - GRPC_ERROR_UNREF(errs[0]); - GRPC_ERROR_UNREF(errs[1]); - gpr_free(addr_str); - return err; } + return err; } /* Return listener at port_index or NULL. Should only be called with s->mu diff --git a/test/core/handshake/client_ssl.c b/test/core/handshake/client_ssl.c index a22133e629..1a06fd6255 100644 --- a/test/core/handshake/client_ssl.c +++ b/test/core/handshake/client_ssl.c @@ -59,13 +59,17 @@ typedef struct { char *alpn_preferred; } server_args; -// From https://wiki.openssl.org/index.php/Simple_TLS_Server. -static int create_socket(int port) { +// Based on https://wiki.openssl.org/index.php/Simple_TLS_Server. +// Pick an arbitrary unused port and return it in *out_port. Return +// an fd>=0 on success. +static int create_socket(int *out_port) { int s; struct sockaddr_in addr; + socklen_t addr_len; + *out_port = -1; addr.sin_family = AF_INET; - addr.sin_port = htons((uint16_t)port); + addr.sin_port = 0; addr.sin_addr.s_addr = htonl(INADDR_ANY); s = socket(AF_INET, SOCK_STREAM, 0); @@ -76,7 +80,7 @@ static int create_socket(int port) { if (bind(s, (struct sockaddr *)&addr, sizeof(addr)) < 0) { perror("Unable to bind"); - gpr_log(GPR_ERROR, "Unable to bind to %d", port); + gpr_log(GPR_ERROR, "%s", "Unable to bind to any port"); close(s); return -1; } @@ -87,6 +91,16 @@ static int create_socket(int port) { return -1; } + addr_len = sizeof(addr); + if (getsockname(s, (struct sockaddr *)&addr, &addr_len) != 0 || + addr_len > sizeof(addr)) { + perror("getsockname"); + gpr_log(GPR_ERROR, "%s", "Unable to get socket local address"); + close(s); + return -1; + } + + *out_port = ntohs(addr.sin_port); return s; } @@ -215,13 +229,12 @@ static bool client_ssl_test(char *server_alpn_preferred) { int server_socket = -1; int socket_retries = 30; while (server_socket == -1 && socket_retries-- > 0) { - port = grpc_pick_unused_port_or_die(); - server_socket = create_socket(port); + server_socket = create_socket(&port); if (server_socket == -1) { sleep(1); } } - GPR_ASSERT(server_socket > 0); + GPR_ASSERT(server_socket > 0 && port > 0); // Launch the TLS server thread. gpr_thd_options thdopt = gpr_thd_options_default(); diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 7122cd1e86..4a6570015f 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -39,9 +39,12 @@ #include "src/core/lib/iomgr/tcp_server.h" #include <errno.h> +#include <ifaddrs.h> #include <netinet/in.h> +#include <stdio.h> #include <string.h> #include <sys/socket.h> +#include <sys/types.h> #include <unistd.h> #include <grpc/grpc.h> @@ -50,6 +53,7 @@ #include <grpc/support/sync.h> #include <grpc/support/time.h> +#include "src/core/lib/iomgr/error.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/lib/iomgr/sockaddr_utils.h" @@ -62,7 +66,7 @@ static gpr_mu *g_mu; static grpc_pollset *g_pollset; static int g_nconnects = 0; -typedef struct on_connect_result { +typedef struct { /* Owns a ref to server. */ grpc_tcp_server *server; unsigned port_index; @@ -70,15 +74,41 @@ typedef struct on_connect_result { int server_fd; } on_connect_result; -typedef struct server_weak_ref { +typedef struct { grpc_tcp_server *server; /* arg is this server_weak_ref. */ grpc_closure server_shutdown; } server_weak_ref; +#define MAX_URI 1024 +typedef struct { + grpc_resolved_address addr; + char str[MAX_URI]; +} test_addr; + +#define MAX_ADDRS 100 +typedef struct { + size_t naddrs; + test_addr addrs[MAX_ADDRS]; +} test_addrs; + static on_connect_result g_result = {NULL, 0, 0, -1}; +static char family_name_buf[1024]; +static const char *sock_family_name(int family) { + if (family == AF_INET) { + return "AF_INET"; + } else if (family == AF_INET6) { + return "AF_INET6"; + } else if (family == AF_UNSPEC) { + return "AF_UNSPEC"; + } else { + sprintf(family_name_buf, "%d", family); + return family_name_buf; + } +} + static void on_connect_result_init(on_connect_result *result) { result->server = NULL; result->port_index = 0; @@ -118,6 +148,18 @@ static void server_weak_ref_set(server_weak_ref *weak_ref, weak_ref->server = server; } +static void test_addr_init_str(test_addr *addr) { + char *str = NULL; + if (grpc_sockaddr_to_string(&str, &addr->addr, 0) != -1) { + size_t str_len; + memcpy(addr->str, str, (str_len = strnlen(str, sizeof(addr->str) - 1))); + addr->str[str_len] = '\0'; + gpr_free(str); + } else { + addr->str[0] = '\0'; + } +} + static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, grpc_pollset *pollset, grpc_tcp_server_acceptor *acceptor) { @@ -168,7 +210,7 @@ static void test_no_op_with_port(void) { memset(&resolved_addr, 0, sizeof(resolved_addr)); resolved_addr.len = sizeof(struct sockaddr_in); addr->sin_family = AF_INET; - int port; + int port = -1; GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr, &port) == GRPC_ERROR_NONE && port > 0); @@ -185,7 +227,7 @@ static void test_no_op_with_port_and_start(void) { GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); LOG_TEST("test_no_op_with_port_and_start"); - int port; + int port = -1; memset(&resolved_addr, 0, sizeof(resolved_addr)); resolved_addr.len = sizeof(struct sockaddr_in); @@ -200,74 +242,115 @@ static void test_no_op_with_port_and_start(void) { grpc_exec_ctx_finish(&exec_ctx); } -static void tcp_connect(grpc_exec_ctx *exec_ctx, const struct sockaddr *remote, - socklen_t remote_len, on_connect_result *result) { +static grpc_error *tcp_connect(grpc_exec_ctx *exec_ctx, const test_addr *remote, + on_connect_result *result) { gpr_timespec deadline = grpc_timeout_seconds_to_deadline(10); - int clifd = socket(remote->sa_family, SOCK_STREAM, 0); + int clifd; int nconnects_before; + const struct sockaddr *remote_addr = + (const struct sockaddr *)remote->addr.addr; + gpr_log(GPR_INFO, "Connecting to %s", remote->str); gpr_mu_lock(g_mu); nconnects_before = g_nconnects; on_connect_result_init(&g_result); - GPR_ASSERT(clifd >= 0); - gpr_log(GPR_DEBUG, "start connect"); - GPR_ASSERT(connect(clifd, remote, remote_len) == 0); + clifd = socket(remote_addr->sa_family, SOCK_STREAM, 0); + if (clifd < 0) { + gpr_mu_unlock(g_mu); + return GRPC_OS_ERROR(errno, "Failed to create socket"); + } + gpr_log(GPR_DEBUG, "start connect to %s", remote->str); + if (connect(clifd, remote_addr, (socklen_t)remote->addr.len) != 0) { + gpr_mu_unlock(g_mu); + close(clifd); + return GRPC_OS_ERROR(errno, "connect"); + } gpr_log(GPR_DEBUG, "wait"); while (g_nconnects == nconnects_before && gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0) { grpc_pollset_worker *worker = NULL; - GPR_ASSERT(GRPC_LOG_IF_ERROR( - "pollset_work", - grpc_pollset_work(exec_ctx, g_pollset, &worker, - gpr_now(GPR_CLOCK_MONOTONIC), deadline))); + grpc_error *err; + if ((err = grpc_pollset_work(exec_ctx, g_pollset, &worker, + gpr_now(GPR_CLOCK_MONOTONIC), deadline)) != + GRPC_ERROR_NONE) { + gpr_mu_unlock(g_mu); + close(clifd); + return err; + } gpr_mu_unlock(g_mu); grpc_exec_ctx_finish(exec_ctx); gpr_mu_lock(g_mu); } gpr_log(GPR_DEBUG, "wait done"); - GPR_ASSERT(g_nconnects == nconnects_before + 1); + if (g_nconnects != nconnects_before + 1) { + gpr_mu_unlock(g_mu); + close(clifd); + return GRPC_ERROR_CREATE("Didn't connect"); + } close(clifd); *result = g_result; gpr_mu_unlock(g_mu); + gpr_log(GPR_INFO, "Result (%d, %d) fd %d", result->port_index, + result->fd_index, result->server_fd); + grpc_tcp_server_unref(exec_ctx, result->server); + return GRPC_ERROR_NONE; } -/* Tests a tcp server with multiple ports. TODO(daniel-j-born): Multiple fds for - the same port should be tested. */ -static void test_connect(unsigned n) { +/* Tests a tcp server on "::" listeners with multiple ports. If channel_args is + non-NULL, pass them to the server. If dst_addrs is non-NULL, use valid addrs + as destination addrs (port is not set). If dst_addrs is NULL, use listener + addrs as destination addrs. If test_dst_addrs is true, test connectivity with + each destination address, set grpc_resolved_address::len=0 for failures, but + don't fail the overall unitest. */ +static void test_connect(size_t num_connects, + const grpc_channel_args *channel_args, + test_addrs *dst_addrs, bool test_dst_addrs) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_resolved_address resolved_addr; grpc_resolved_address resolved_addr1; - struct sockaddr_storage *addr = (struct sockaddr_storage *)resolved_addr.addr; - struct sockaddr_storage *addr1 = + struct sockaddr_storage *const addr = + (struct sockaddr_storage *)resolved_addr.addr; + struct sockaddr_storage *const addr1 = (struct sockaddr_storage *)resolved_addr1.addr; unsigned svr_fd_count; + int port; int svr_port; unsigned svr1_fd_count; int svr1_port; grpc_tcp_server *s; + const unsigned num_ports = 2; GPR_ASSERT(GRPC_ERROR_NONE == - grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); - unsigned i; + grpc_tcp_server_create(&exec_ctx, NULL, channel_args, &s)); + unsigned port_num; server_weak_ref weak_ref; server_weak_ref_init(&weak_ref); + server_weak_ref_set(&weak_ref, s); LOG_TEST("test_connect"); - gpr_log(GPR_INFO, "clients=%d", n); + gpr_log(GPR_INFO, + "clients=%lu, num chan args=%lu, remote IP=%s, test_dst_addrs=%d", + (unsigned long)num_connects, + (unsigned long)(channel_args != NULL ? channel_args->num_args : 0), + dst_addrs != NULL ? "<specific>" : "::", test_dst_addrs); memset(&resolved_addr, 0, sizeof(resolved_addr)); memset(&resolved_addr1, 0, sizeof(resolved_addr1)); resolved_addr.len = sizeof(struct sockaddr_storage); resolved_addr1.len = sizeof(struct sockaddr_storage); addr->ss_family = addr1->ss_family = AF_INET; - GPR_ASSERT(GRPC_ERROR_NONE == - grpc_tcp_server_add_port(s, &resolved_addr, &svr_port)); + GPR_ASSERT(GRPC_LOG_IF_ERROR( + "grpc_tcp_server_add_port", + grpc_tcp_server_add_port(s, &resolved_addr, &svr_port))); + gpr_log(GPR_INFO, "Allocated port %d", svr_port); GPR_ASSERT(svr_port > 0); /* Cannot use wildcard (port==0), because add_port() will try to reuse the same port as a previous add_port(). */ svr1_port = grpc_pick_unused_port_or_die(); + GPR_ASSERT(svr1_port > 0); + gpr_log(GPR_INFO, "Picked unused port %d", svr1_port); grpc_sockaddr_set_port(&resolved_addr1, svr1_port); - GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr1, &svr_port) == + GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr1, &port) == GRPC_ERROR_NONE && - svr_port == svr1_port); + port == svr1_port); /* Bad port_index. */ GPR_ASSERT(grpc_tcp_server_port_fd_count(s, 2) == 0); @@ -283,58 +366,70 @@ static void test_connect(unsigned n) { svr1_fd_count = grpc_tcp_server_port_fd_count(s, 1); GPR_ASSERT(svr1_fd_count >= 1); - for (i = 0; i < svr_fd_count; ++i) { - int fd = grpc_tcp_server_port_fd(s, 0, i); - GPR_ASSERT(fd >= 0); - if (i == 0) { - GPR_ASSERT(getsockname(fd, (struct sockaddr *)addr, - (socklen_t *)&resolved_addr.len) == 0); - GPR_ASSERT(resolved_addr.len <= sizeof(*addr)); - } - } - for (i = 0; i < svr1_fd_count; ++i) { - int fd = grpc_tcp_server_port_fd(s, 1, i); - GPR_ASSERT(fd >= 0); - if (i == 0) { - GPR_ASSERT(getsockname(fd, (struct sockaddr *)addr1, - (socklen_t *)&resolved_addr1.len) == 0); - GPR_ASSERT(resolved_addr1.len <= sizeof(*addr1)); - } - } - grpc_tcp_server_start(&exec_ctx, s, &g_pollset, 1, on_connect, NULL); - for (i = 0; i < n; i++) { - on_connect_result result; - int svr_fd; - on_connect_result_init(&result); - tcp_connect(&exec_ctx, (struct sockaddr *)addr, - (socklen_t)resolved_addr.len, &result); - GPR_ASSERT(result.server_fd >= 0); - svr_fd = result.server_fd; - GPR_ASSERT(grpc_tcp_server_port_fd(s, result.port_index, result.fd_index) == - result.server_fd); - GPR_ASSERT(result.port_index == 0); - GPR_ASSERT(result.fd_index < svr_fd_count); - GPR_ASSERT(result.server == s); - if (weak_ref.server == NULL) { - server_weak_ref_set(&weak_ref, result.server); + if (dst_addrs != NULL) { + int ports[] = {svr_port, svr1_port}; + for (port_num = 0; port_num < num_ports; ++port_num) { + size_t dst_idx; + size_t num_tested = 0; + for (dst_idx = 0; dst_idx < dst_addrs->naddrs; ++dst_idx) { + test_addr dst = dst_addrs->addrs[dst_idx]; + on_connect_result result; + grpc_error *err; + if (dst.addr.len == 0) { + gpr_log(GPR_DEBUG, "Skipping test of non-functional local IP %s", + dst.str); + continue; + } + GPR_ASSERT(grpc_sockaddr_set_port(&dst.addr, ports[port_num])); + test_addr_init_str(&dst); + ++num_tested; + on_connect_result_init(&result); + if ((err = tcp_connect(&exec_ctx, &dst, &result)) == GRPC_ERROR_NONE && + result.server_fd >= 0 && result.server == s) { + continue; + } + gpr_log(GPR_ERROR, "Failed to connect to %s: %s", dst.str, + grpc_error_string(err)); + GPR_ASSERT(test_dst_addrs); + dst_addrs->addrs[dst_idx].addr.len = 0; + GRPC_ERROR_UNREF(err); + } + GPR_ASSERT(num_tested > 0); + } + } else { + for (port_num = 0; port_num < num_ports; ++port_num) { + const unsigned num_fds = grpc_tcp_server_port_fd_count(s, port_num); + unsigned fd_num; + for (fd_num = 0; fd_num < num_fds; ++fd_num) { + int fd = grpc_tcp_server_port_fd(s, port_num, fd_num); + size_t connect_num; + test_addr dst; + GPR_ASSERT(fd >= 0); + dst.addr.len = sizeof(dst.addr.addr); + GPR_ASSERT(getsockname(fd, (struct sockaddr *)dst.addr.addr, + (socklen_t *)&dst.addr.len) == 0); + GPR_ASSERT(dst.addr.len <= sizeof(dst.addr.addr)); + test_addr_init_str(&dst); + gpr_log(GPR_INFO, "(%d, %d) fd %d family %s listening on %s", port_num, + fd_num, fd, sock_family_name(addr->ss_family), dst.str); + for (connect_num = 0; connect_num < num_connects; ++connect_num) { + on_connect_result result; + on_connect_result_init(&result); + GPR_ASSERT(GRPC_LOG_IF_ERROR("tcp_connect", + tcp_connect(&exec_ctx, &dst, &result))); + GPR_ASSERT(result.server_fd == fd); + GPR_ASSERT(result.port_index == port_num); + GPR_ASSERT(result.fd_index == fd_num); + GPR_ASSERT(result.server == s); + GPR_ASSERT( + grpc_tcp_server_port_fd(s, result.port_index, result.fd_index) == + result.server_fd); + } + } } - grpc_tcp_server_unref(&exec_ctx, result.server); - - on_connect_result_init(&result); - tcp_connect(&exec_ctx, (struct sockaddr *)addr1, - (socklen_t)resolved_addr1.len, &result); - GPR_ASSERT(result.server_fd >= 0); - GPR_ASSERT(result.server_fd != svr_fd); - GPR_ASSERT(grpc_tcp_server_port_fd(s, result.port_index, result.fd_index) == - result.server_fd); - GPR_ASSERT(result.port_index == 1); - GPR_ASSERT(result.fd_index < svr_fd_count); - GPR_ASSERT(result.server == s); - grpc_tcp_server_unref(&exec_ctx, result.server); } - /* Weak ref to server valid until final unref. */ GPR_ASSERT(weak_ref.server != NULL); GPR_ASSERT(grpc_tcp_server_port_fd(s, 0, 0) >= 0); @@ -354,6 +449,12 @@ static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, int main(int argc, char **argv) { grpc_closure destroyed; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_arg chan_args[] = { + {GRPC_ARG_INTEGER, GRPC_ARG_EXPAND_WILDCARD_ADDRS, {.integer = 1}}}; + const grpc_channel_args channel_args = {1, chan_args}; + struct ifaddrs *ifa = NULL; + struct ifaddrs *ifa_it; + test_addrs dst_addrs; grpc_test_init(argc, argv); grpc_init(); g_pollset = gpr_malloc(grpc_pollset_size()); @@ -363,8 +464,45 @@ int main(int argc, char **argv) { test_no_op_with_start(); test_no_op_with_port(); test_no_op_with_port_and_start(); - test_connect(1); - test_connect(10); + + if (getifaddrs(&ifa) != 0 || ifa == NULL) { + gpr_log(GPR_ERROR, "getifaddrs: %s", strerror(errno)); + return EXIT_FAILURE; + } + dst_addrs.naddrs = 0; + for (ifa_it = ifa; ifa_it != NULL && dst_addrs.naddrs < MAX_ADDRS; + ifa_it = ifa_it->ifa_next) { + if (ifa_it->ifa_addr == NULL) { + continue; + } else if (ifa_it->ifa_addr->sa_family == AF_INET) { + dst_addrs.addrs[dst_addrs.naddrs].addr.len = sizeof(struct sockaddr_in); + } else if (ifa_it->ifa_addr->sa_family == AF_INET6) { + dst_addrs.addrs[dst_addrs.naddrs].addr.len = sizeof(struct sockaddr_in6); + } else { + continue; + } + memcpy(dst_addrs.addrs[dst_addrs.naddrs].addr.addr, ifa_it->ifa_addr, + dst_addrs.addrs[dst_addrs.naddrs].addr.len); + GPR_ASSERT( + grpc_sockaddr_set_port(&dst_addrs.addrs[dst_addrs.naddrs].addr, 0)); + test_addr_init_str(&dst_addrs.addrs[dst_addrs.naddrs]); + ++dst_addrs.naddrs; + } + freeifaddrs(ifa); + ifa = NULL; + + /* Connect to same addresses as listeners. */ + test_connect(1, NULL, NULL, false); + test_connect(10, NULL, NULL, false); + + /* Set dst_addrs.addrs[i].len=0 for dst_addrs that are unreachable with a "::" + listener. */ + test_connect(1, NULL, &dst_addrs, true); + + /* Test connect(2) with dst_addrs. */ + test_connect(1, &channel_args, &dst_addrs, false); + /* Test connect(2) with dst_addrs. */ + test_connect(10, &channel_args, &dst_addrs, false); grpc_closure_init(&destroyed, destroy_pollset, g_pollset, grpc_schedule_on_exec_ctx); @@ -372,7 +510,7 @@ int main(int argc, char **argv) { grpc_exec_ctx_finish(&exec_ctx); grpc_shutdown(); gpr_free(g_pollset); - return 0; + return EXIT_SUCCESS; } #else /* GRPC_POSIX_SOCKET */ |