diff options
-rw-r--r-- | src/core/client_config/connector.c | 4 | ||||
-rw-r--r-- | src/core/client_config/connector.h | 6 | ||||
-rw-r--r-- | src/core/client_config/subchannel.c | 4 | ||||
-rw-r--r-- | src/core/client_config/subchannel.h | 1 | ||||
-rw-r--r-- | src/core/iomgr/fd_posix.c | 3 | ||||
-rw-r--r-- | src/core/iomgr/iomgr.c | 10 | ||||
-rw-r--r-- | src/core/iomgr/pollset_multipoller_with_epoll.c | 2 | ||||
-rw-r--r-- | src/core/support/string.c | 2 | ||||
-rw-r--r-- | src/core/surface/channel_create.c | 4 | ||||
-rw-r--r-- | src/core/surface/secure_channel_create.c | 36 | ||||
-rw-r--r-- | src/core/transport/chttp2/parsing.c | 4 | ||||
-rw-r--r-- | test/core/util/port_posix.c | 92 | ||||
-rwxr-xr-x | tools/run_tests/port_server.py | 30 |
13 files changed, 174 insertions, 24 deletions
diff --git a/src/core/client_config/connector.c b/src/core/client_config/connector.c index a8cd5fc149..c1e583e4a5 100644 --- a/src/core/client_config/connector.c +++ b/src/core/client_config/connector.c @@ -47,3 +47,7 @@ void grpc_connector_connect(grpc_connector *connector, grpc_iomgr_closure *notify) { connector->vtable->connect(connector, in_args, out_args, notify); } + +void grpc_connector_shutdown(grpc_connector *connector) { + connector->vtable->shutdown(connector); +} diff --git a/src/core/client_config/connector.h b/src/core/client_config/connector.h index 39f3467990..01aa716412 100644 --- a/src/core/client_config/connector.h +++ b/src/core/client_config/connector.h @@ -70,6 +70,9 @@ typedef struct { struct grpc_connector_vtable { void (*ref)(grpc_connector *connector); void (*unref)(grpc_connector *connector); + /** Implementation of grpc_connector_shutdown */ + void (*shutdown)(grpc_connector *connector); + /** Implementation of grpc_connector_connect */ void (*connect)(grpc_connector *connector, const grpc_connect_in_args *in_args, grpc_connect_out_args *out_args, grpc_iomgr_closure *notify); @@ -77,9 +80,12 @@ struct grpc_connector_vtable { void grpc_connector_ref(grpc_connector *connector); void grpc_connector_unref(grpc_connector *connector); +/** Connect using the connector: max one outstanding call at a time */ void grpc_connector_connect(grpc_connector *connector, const grpc_connect_in_args *in_args, grpc_connect_out_args *out_args, grpc_iomgr_closure *notify); +/** Cancel any pending connection */ +void grpc_connector_shutdown(grpc_connector *connector); #endif diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c index ca52c75beb..876d2aa418 100644 --- a/src/core/client_config/subchannel.c +++ b/src/core/client_config/subchannel.c @@ -439,6 +439,10 @@ void grpc_subchannel_process_transport_op(grpc_subchannel *c, if (cancel_alarm) { grpc_alarm_cancel(&c->alarm); } + + if (op->disconnect) { + grpc_connector_shutdown(c->connector); + } } static void on_state_changed(void *p, int iomgr_success) { diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h index 2e36c69134..29a0b67d2d 100644 --- a/src/core/client_config/subchannel.h +++ b/src/core/client_config/subchannel.h @@ -43,6 +43,7 @@ typedef struct grpc_subchannel grpc_subchannel; typedef struct grpc_subchannel_call grpc_subchannel_call; typedef struct grpc_subchannel_args grpc_subchannel_args; +#define GRPC_SUBCHANNEL_REFCOUNT_DEBUG #ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG #define GRPC_SUBCHANNEL_REF(p, r) \ grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c index 2d08a77a70..38a543e36e 100644 --- a/src/core/iomgr/fd_posix.c +++ b/src/core/iomgr/fd_posix.c @@ -213,10 +213,9 @@ void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_closure *on_done, const char *reason) { fd->on_done_closure = on_done; shutdown(fd->fd, SHUT_RDWR); - REF_BY(fd, 1, reason); /* remove active status, but keep referenced */ gpr_mu_lock(&fd->watcher_mu); + REF_BY(fd, 1, reason); /* remove active status, but keep referenced */ if (!has_watchers(fd)) { - GPR_ASSERT(!fd->closed); fd->closed = 1; close(fd->fd); if (fd->on_done_closure) { diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index 1dd03992ae..d6ca5d1f71 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -34,16 +34,18 @@ #include "src/core/iomgr/iomgr.h" #include <stdlib.h> +#include <string.h> -#include "src/core/iomgr/iomgr_internal.h" -#include "src/core/iomgr/alarm_internal.h" -#include "src/core/support/string.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/thd.h> +#include "src/core/iomgr/iomgr_internal.h" +#include "src/core/iomgr/alarm_internal.h" +#include "src/core/support/string.h" + static gpr_mu g_mu; static gpr_cv g_rcv; static grpc_iomgr_closure *g_cbs_head = NULL; @@ -179,6 +181,8 @@ void grpc_iomgr_shutdown(void) { } gpr_mu_unlock(&g_mu); + memset(&g_root_object, 0, sizeof(g_root_object)); + grpc_kick_poller(); gpr_event_wait(&g_background_callback_executor_done, gpr_inf_future(GPR_CLOCK_REALTIME)); diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index 8f62ce2954..481bdc4ede 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -72,7 +72,7 @@ static void finally_add_fd(grpc_pollset *pollset, grpc_fd *fd) { to this pollset whilst adding, but that should be benign. */ GPR_ASSERT(grpc_fd_begin_poll(fd, pollset, 0, 0, &watcher) == 0); if (watcher.fd != NULL) { - ev.events = EPOLLIN | EPOLLOUT | EPOLLET; + ev.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET); ev.data.ptr = fd; err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, fd->fd, &ev); if (err < 0) { diff --git a/src/core/support/string.c b/src/core/support/string.c index af0389ea83..e0ffeb8a4a 100644 --- a/src/core/support/string.c +++ b/src/core/support/string.c @@ -101,7 +101,7 @@ static void asciidump(dump_out *out, const char *buf, size_t len) { dump_out_append(out, '\''); } for (cur = beg; cur != end; ++cur) { - dump_out_append(out, isprint(*cur) ? *(char *)cur : '.'); + dump_out_append(out, (char)(isprint(*cur) ? *(char *)cur : '.')); } if (!out_was_empty) { dump_out_append(out, '\''); diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 9e2cf1cf66..d323d0d74f 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -88,6 +88,8 @@ static void connected(void *arg, grpc_endpoint *tcp) { grpc_iomgr_add_callback(notify); } +static void connector_shutdown(grpc_connector *con) {} + static void connector_connect(grpc_connector *con, const grpc_connect_in_args *args, grpc_connect_out_args *result, @@ -103,7 +105,7 @@ static void connector_connect(grpc_connector *con, } static const grpc_connector_vtable connector_vtable = { - connector_ref, connector_unref, connector_connect}; + connector_ref, connector_unref, connector_shutdown, connector_connect}; typedef struct { grpc_subchannel_factory base; diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index 9b554eeb70..52c5e93988 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -61,6 +61,9 @@ typedef struct { grpc_iomgr_closure *notify; grpc_connect_in_args args; grpc_connect_out_args *result; + + gpr_mu mu; + grpc_endpoint *connecting_endpoint; } connector; static void connector_ref(grpc_connector *con) { @@ -81,10 +84,20 @@ static void on_secure_transport_setup_done(void *arg, grpc_endpoint *secure_endpoint) { connector *c = arg; grpc_iomgr_closure *notify; - if (status != GRPC_SECURITY_OK) { + gpr_mu_lock(&c->mu); + if (c->connecting_endpoint == NULL) { + memset(c->result, 0, sizeof(*c->result)); + gpr_mu_unlock(&c->mu); + } else if (status != GRPC_SECURITY_OK) { + GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint); gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status); memset(c->result, 0, sizeof(*c->result)); + c->connecting_endpoint = NULL; + gpr_mu_unlock(&c->mu); } else { + GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint); + c->connecting_endpoint = NULL; + gpr_mu_unlock(&c->mu); c->result->transport = grpc_create_chttp2_transport( c->args.channel_args, secure_endpoint, c->args.metadata_context, 1); grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0); @@ -102,6 +115,10 @@ static void connected(void *arg, grpc_endpoint *tcp) { connector *c = arg; grpc_iomgr_closure *notify; if (tcp != NULL) { + gpr_mu_lock(&c->mu); + GPR_ASSERT(c->connecting_endpoint == NULL); + c->connecting_endpoint = tcp; + gpr_mu_unlock(&c->mu); grpc_setup_secure_transport(&c->security_connector->base, tcp, on_secure_transport_setup_done, c); } else { @@ -112,6 +129,18 @@ static void connected(void *arg, grpc_endpoint *tcp) { } } +static void connector_shutdown(grpc_connector *con) { + connector *c = (connector *)con; + grpc_endpoint *ep; + gpr_mu_lock(&c->mu); + ep = c->connecting_endpoint; + c->connecting_endpoint = NULL; + gpr_mu_unlock(&c->mu); + if (ep) { + grpc_endpoint_shutdown(ep); + } +} + static void connector_connect(grpc_connector *con, const grpc_connect_in_args *args, grpc_connect_out_args *result, @@ -122,12 +151,15 @@ static void connector_connect(grpc_connector *con, c->notify = notify; c->args = *args; c->result = result; + gpr_mu_lock(&c->mu); + GPR_ASSERT(c->connecting_endpoint == NULL); + gpr_mu_unlock(&c->mu); grpc_tcp_client_connect(connected, c, args->interested_parties, args->addr, args->addr_len, args->deadline); } static const grpc_connector_vtable connector_vtable = { - connector_ref, connector_unref, connector_connect}; + connector_ref, connector_unref, connector_shutdown, connector_connect}; typedef struct { grpc_subchannel_factory base; diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index a592ce7d28..f26f446787 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -486,7 +486,7 @@ static int init_skip_frame_parser( transport_parsing->hpack_parser.on_header_user_data = NULL; transport_parsing->hpack_parser.is_boundary = is_eoh; transport_parsing->hpack_parser.is_eof = - is_eoh ? transport_parsing->header_eof : 0; + (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0); } else { transport_parsing->parser = skip_parser; } @@ -696,7 +696,7 @@ static int init_header_frame_parser( transport_parsing->hpack_parser.on_header_user_data = transport_parsing; transport_parsing->hpack_parser.is_boundary = is_eoh; transport_parsing->hpack_parser.is_eof = - is_eoh ? transport_parsing->header_eof : 0; + (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0); if (!is_continuation && (transport_parsing->incoming_frame_flags & GRPC_CHTTP2_FLAG_HAS_PRIORITY)) { grpc_chttp2_hpack_parser_set_has_priority(&transport_parsing->hpack_parser); diff --git a/test/core/util/port_posix.c b/test/core/util/port_posix.c index be45bae496..03f4c4912e 100644 --- a/test/core/util/port_posix.c +++ b/test/core/util/port_posix.c @@ -47,6 +47,7 @@ #include <grpc/grpc.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> #include "src/core/httpcli/httpcli.h" #include "src/core/support/env.h" @@ -66,7 +67,70 @@ static int has_port_been_chosen(int port) { return 0; } -static void free_chosen_ports() { gpr_free(chosen_ports); } +typedef struct freereq { + grpc_pollset pollset; + int done; +} freereq; + +static void destroy_pollset_and_shutdown(void *p) { + grpc_pollset_destroy(p); + grpc_shutdown(); +} + +static void freed_port_from_server(void *arg, + const grpc_httpcli_response *response) { + freereq *pr = arg; + gpr_mu_lock(GRPC_POLLSET_MU(&pr->pollset)); + pr->done = 1; + grpc_pollset_kick(&pr->pollset, NULL); + gpr_mu_unlock(GRPC_POLLSET_MU(&pr->pollset)); +} + +static void free_port_using_server(char *server, int port) { + grpc_httpcli_context context; + grpc_httpcli_request req; + freereq pr; + char *path; + + grpc_init(); + + memset(&pr, 0, sizeof(pr)); + memset(&req, 0, sizeof(req)); + grpc_pollset_init(&pr.pollset); + + req.host = server; + gpr_asprintf(&path, "/drop/%d", port); + req.path = path; + + grpc_httpcli_context_init(&context); + grpc_httpcli_get(&context, &pr.pollset, &req, + GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), freed_port_from_server, + &pr); + gpr_mu_lock(GRPC_POLLSET_MU(&pr.pollset)); + while (!pr.done) { + grpc_pollset_worker worker; + grpc_pollset_work(&pr.pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC), + GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1)); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&pr.pollset)); + + grpc_httpcli_context_destroy(&context); + grpc_pollset_shutdown(&pr.pollset, destroy_pollset_and_shutdown, &pr.pollset); + gpr_free(path); +} + +static void free_chosen_ports() { + char *env = gpr_getenv("GRPC_TEST_PORT_SERVER"); + if (env != NULL) { + size_t i; + for (i = 0; i < num_chosen_ports; i++) { + free_port_using_server(env, chosen_ports[i]); + } + gpr_free(env); + } + + gpr_free(chosen_ports); +} static void chose_port(int port) { if (chosen_ports == NULL) { @@ -131,6 +195,9 @@ static int is_port_available(int *port, int is_tcp) { typedef struct portreq { grpc_pollset pollset; int port; + int retries; + char *server; + grpc_httpcli_context *ctx; } portreq; static void got_port_from_server(void *arg, @@ -138,6 +205,19 @@ static void got_port_from_server(void *arg, size_t i; int port = 0; portreq *pr = arg; + if (!response || response->status != 200) { + grpc_httpcli_request req; + memset(&req, 0, sizeof(req)); + GPR_ASSERT(pr->retries < 10); + pr->retries++; + req.host = pr->server; + req.path = "/get"; + gpr_log(GPR_DEBUG, "failed port pick from server: retrying"); + sleep(1); + grpc_httpcli_get(pr->ctx, &pr->pollset, &req, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), + got_port_from_server, pr); + return; + } GPR_ASSERT(response); GPR_ASSERT(response->status == 200); for (i = 0; i < response->body_length; i++) { @@ -151,11 +231,6 @@ static void got_port_from_server(void *arg, gpr_mu_unlock(GRPC_POLLSET_MU(&pr->pollset)); } -static void destroy_pollset_and_shutdown(void *p) { - grpc_pollset_destroy(p); - grpc_shutdown(); -} - static int pick_port_using_server(char *server) { grpc_httpcli_context context; grpc_httpcli_request req; @@ -167,6 +242,8 @@ static int pick_port_using_server(char *server) { memset(&req, 0, sizeof(req)); grpc_pollset_init(&pr.pollset); pr.port = -1; + pr.server = server; + pr.ctx = &context; req.host = server; req.path = "/get"; @@ -211,8 +288,9 @@ int grpc_pick_unused_port(void) { int port = pick_port_using_server(env); gpr_free(env); if (port != 0) { - return port; + chose_port(port); } + return port; } for (;;) { diff --git a/tools/run_tests/port_server.py b/tools/run_tests/port_server.py index 0f81470d28..d14c829fe0 100755 --- a/tools/run_tests/port_server.py +++ b/tools/run_tests/port_server.py @@ -37,6 +37,7 @@ import os import socket import sys import time +import yaml argp = argparse.ArgumentParser(description='Server for httpcli_test') argp.add_argument('-p', '--port', default=12345, type=int) @@ -51,16 +52,17 @@ with open(__file__) as f: _MY_VERSION = hashlib.sha1(f.read()).hexdigest() -def refill_pool(): +def refill_pool(max_timeout): """Scan for ports not marked for being in use""" - for i in range(10000, 65000): + for i in range(1025, 32767): if len(pool) > 100: break if i in in_use: age = time.time() - in_use[i] - if age < 600: + if age < max_timeout: continue del in_use[i] s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: s.bind(('localhost', i)) pool.append(i) @@ -73,8 +75,12 @@ def refill_pool(): def allocate_port(): global pool global in_use - if not pool: - refill_pool() + max_timeout = 600 + while not pool: + refill_pool(max_timeout) + if not pool: + time.sleep(1) + max_timeout /= 2 port = pool[0] pool = pool[1:] in_use[port] = time.time() @@ -97,12 +103,26 @@ class Handler(BaseHTTPServer.BaseHTTPRequestHandler): p = allocate_port() self.log_message('allocated port %d' % p) self.wfile.write('%d' % p) + elif self.path[0:6] == '/drop/': + self.send_response(200) + self.send_header('Content-Type', 'text/plain') + self.end_headers() + p = int(self.path[6:]) + del in_use[p] + pool.append(p) + self.log_message('drop port %d' % p) elif self.path == '/version': # fetch a version string and the current process pid self.send_response(200) self.send_header('Content-Type', 'text/plain') self.end_headers() self.wfile.write(_MY_VERSION) + elif self.path == '/dump': + self.send_response(200) + self.send_header('Content-Type', 'text/plain') + self.end_headers() + now = time.time() + self.wfile.write(yaml.dump({'pool': pool, 'in_use': dict((k, now - v) for k, v in in_use.iteritems())})) elif self.path == '/quit': self.send_response(200) self.end_headers() |