From 15521de93fc0a9feb2e1a555d249d8fd070ec9ab Mon Sep 17 00:00:00 2001 From: Yuchen Zeng Date: Thu, 17 Nov 2016 20:39:27 -0800 Subject: [PATCH] Request a pollset_set in grpc_resolve_address --- .../ext/resolver/dns/native/dns_resolver.c | 1 + src/core/lib/http/httpcli.c | 1 + src/core/lib/iomgr/resolve_address.h | 2 + src/core/lib/iomgr/resolve_address_posix.c | 9 ++- src/core/lib/iomgr/resolve_address_uv.c | 9 ++- test/core/end2end/fuzzers/api_fuzzer.c | 4 +- test/core/iomgr/resolve_address_test.c | 74 ++++++++++++++++--- 7 files changed, 82 insertions(+), 18 deletions(-) diff --git a/src/core/ext/resolver/dns/native/dns_resolver.c b/src/core/ext/resolver/dns/native/dns_resolver.c index a7392e14ad7..688c4fa845d 100644 --- a/src/core/ext/resolver/dns/native/dns_resolver.c +++ b/src/core/ext/resolver/dns/native/dns_resolver.c @@ -218,6 +218,7 @@ static void dns_start_resolving_locked(grpc_exec_ctx *exec_ctx, r->resolving = true; r->addresses = NULL; grpc_resolve_address(exec_ctx, r->name_to_resolve, r->default_port, + r->base.pollset_set, grpc_closure_create(dns_on_resolved, r), &r->addresses); } diff --git a/src/core/lib/http/httpcli.c b/src/core/lib/http/httpcli.c index fdb8abaa2df..1035f311095 100644 --- a/src/core/lib/http/httpcli.c +++ b/src/core/lib/http/httpcli.c @@ -278,6 +278,7 @@ static void internal_request_begin(grpc_exec_ctx *exec_ctx, grpc_polling_entity_add_to_pollset_set(exec_ctx, req->pollent, req->context->pollset_set); grpc_resolve_address(exec_ctx, request->host, req->handshaker->default_port, + req->context->pollset_set, grpc_closure_create(on_resolved, req), &req->addresses); } diff --git a/src/core/lib/iomgr/resolve_address.h b/src/core/lib/iomgr/resolve_address.h index 275924448a1..e03d16fa4ea 100644 --- a/src/core/lib/iomgr/resolve_address.h +++ b/src/core/lib/iomgr/resolve_address.h @@ -36,6 +36,7 @@ #include #include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/iomgr/pollset_set.h" #define GRPC_MAX_SOCKADDR_SIZE 128 @@ -54,6 +55,7 @@ typedef struct { /* TODO(ctiller): add a timeout here */ extern void (*grpc_resolve_address)(grpc_exec_ctx *exec_ctx, const char *addr, const char *default_port, + grpc_pollset_set *interested_parties, grpc_closure *on_done, grpc_resolved_addresses **addresses); /* Destroy resolved addresses */ diff --git a/src/core/lib/iomgr/resolve_address_posix.c b/src/core/lib/iomgr/resolve_address_posix.c index de791b2b67c..821932e562d 100644 --- a/src/core/lib/iomgr/resolve_address_posix.c +++ b/src/core/lib/iomgr/resolve_address_posix.c @@ -181,6 +181,7 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) { static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, const char *default_port, + grpc_pollset_set *interested_parties, grpc_closure *on_done, grpc_resolved_addresses **addrs) { request *r = gpr_malloc(sizeof(request)); @@ -192,9 +193,9 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, grpc_executor_push(&r->request_closure, GRPC_ERROR_NONE); } -void (*grpc_resolve_address)(grpc_exec_ctx *exec_ctx, const char *name, - const char *default_port, grpc_closure *on_done, - grpc_resolved_addresses **addrs) = - resolve_address_impl; +void (*grpc_resolve_address)( + grpc_exec_ctx *exec_ctx, const char *name, const char *default_port, + grpc_pollset_set *interested_parties, grpc_closure *on_done, + grpc_resolved_addresses **addrs) = resolve_address_impl; #endif diff --git a/src/core/lib/iomgr/resolve_address_uv.c b/src/core/lib/iomgr/resolve_address_uv.c index b8295acfa14..3269c4f09ff 100644 --- a/src/core/lib/iomgr/resolve_address_uv.c +++ b/src/core/lib/iomgr/resolve_address_uv.c @@ -181,6 +181,7 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) { static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, const char *default_port, + grpc_pollset_set *interested_parties, grpc_closure *on_done, grpc_resolved_addresses **addrs) { uv_getaddrinfo_t *req; @@ -223,9 +224,9 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, } } -void (*grpc_resolve_address)(grpc_exec_ctx *exec_ctx, const char *name, - const char *default_port, grpc_closure *on_done, - grpc_resolved_addresses **addrs) = - resolve_address_impl; +void (*grpc_resolve_address)( + grpc_exec_ctx *exec_ctx, const char *name, const char *default_port, + grpc_pollset_set *interested_parties, grpc_closure *on_done, + grpc_resolved_addresses **addrs) = resolve_address_impl; #endif /* GRPC_UV */ diff --git a/test/core/end2end/fuzzers/api_fuzzer.c b/test/core/end2end/fuzzers/api_fuzzer.c index 19ac6ced14b..746134c85be 100644 --- a/test/core/end2end/fuzzers/api_fuzzer.c +++ b/test/core/end2end/fuzzers/api_fuzzer.c @@ -361,7 +361,9 @@ static void finish_resolve(grpc_exec_ctx *exec_ctx, void *arg, } void my_resolve_address(grpc_exec_ctx *exec_ctx, const char *addr, - const char *default_port, grpc_closure *on_done, + const char *default_port, + grpc_pollset_set *interested_parties, + grpc_closure *on_done, grpc_resolved_addresses **addresses) { addr_req *r = gpr_malloc(sizeof(*r)); r->addr = gpr_strdup(addr); diff --git a/test/core/iomgr/resolve_address_test.c b/test/core/iomgr/resolve_address_test.c index 2dd0d88b3f8..2f533137d83 100644 --- a/test/core/iomgr/resolve_address_test.c +++ b/test/core/iomgr/resolve_address_test.c @@ -34,6 +34,7 @@ #include "src/core/lib/iomgr/resolve_address.h" #include #include +#include #include #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/iomgr.h" @@ -46,16 +47,64 @@ static gpr_timespec test_deadline(void) { typedef struct args_struct { gpr_event ev; grpc_resolved_addresses *addrs; + gpr_atm done_atm; + gpr_mu *mu; + grpc_pollset *pollset; + grpc_pollset_set *pollset_set; } args_struct; void args_init(args_struct *args) { gpr_event_init(&args->ev); + grpc_pollset_init(args->pollset, &args->mu); + args->pollset_set = grpc_pollset_set_create(); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_pollset_set_add_pollset(&exec_ctx, args->pollset_set, args->pollset); + grpc_exec_ctx_finish(&exec_ctx); args->addrs = NULL; } void args_finish(args_struct *args) { GPR_ASSERT(gpr_event_wait(&args->ev, test_deadline())); grpc_resolved_addresses_destroy(args->addrs); + grpc_pollset_set_destroy(args->pollset_set); + grpc_pollset_destroy(args->pollset); +} + +static gpr_timespec n_sec_deadline(int seconds) { + return gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_seconds(seconds, GPR_TIMESPAN)); +} + +static void actually_poll(void *argsp) { + args_struct *args = argsp; + gpr_timespec deadline = n_sec_deadline(10); + grpc_pollset_worker *worker = NULL; + while (true) { + bool done = gpr_atm_acq_load(&args->done_atm) != 0; + if (done) { + break; + } + gpr_timespec time_left = + gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)); + gpr_log(GPR_DEBUG, "done=%d, time_left=%" PRId64 ".%09d", done, + time_left.tv_sec, time_left.tv_nsec); + GPR_ASSERT(gpr_time_cmp(time_left, gpr_time_0(GPR_TIMESPAN)) >= 0); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_mu_lock(args->mu); + GRPC_LOG_IF_ERROR( + "pollset_work", + grpc_pollset_work(&exec_ctx, args->pollset, &worker, + gpr_now(GPR_CLOCK_REALTIME), n_sec_deadline(1))); + gpr_mu_unlock(args->mu); + grpc_exec_ctx_finish(&exec_ctx); + } + gpr_event_set(&args->ev, (void *)1); +} + +static void poll_pollset_until_request_done(args_struct *args) { + gpr_atm_rel_store(&args->done_atm, 0); + gpr_thd_id id; + gpr_thd_new(&id, actually_poll, args, NULL); } static void must_succeed(grpc_exec_ctx *exec_ctx, void *argsp, @@ -64,20 +113,21 @@ static void must_succeed(grpc_exec_ctx *exec_ctx, void *argsp, GPR_ASSERT(err == GRPC_ERROR_NONE); GPR_ASSERT(args->addrs != NULL); GPR_ASSERT(args->addrs->naddrs > 0); - gpr_event_set(&args->ev, (void *)1); + gpr_atm_rel_store(&args->done_atm, 1); } static void must_fail(grpc_exec_ctx *exec_ctx, void *argsp, grpc_error *err) { args_struct *args = argsp; GPR_ASSERT(err != GRPC_ERROR_NONE); - gpr_event_set(&args->ev, (void *)1); + gpr_atm_rel_store(&args->done_atm, 1); } static void test_localhost(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, "localhost:1", NULL, + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, "localhost:1", NULL, args.pollset_set, grpc_closure_create(must_succeed, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -87,7 +137,8 @@ static void test_default_port(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, "localhost", "1", + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, "localhost", "1", args.pollset_set, grpc_closure_create(must_succeed, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -97,7 +148,8 @@ static void test_missing_default_port(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, "localhost", NULL, + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, "localhost", NULL, args.pollset_set, grpc_closure_create(must_fail, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -107,7 +159,8 @@ static void test_ipv6_with_port(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, "[2001:db8::1]:1", NULL, + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, "[2001:db8::1]:1", NULL, args.pollset_set, grpc_closure_create(must_succeed, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -122,7 +175,8 @@ static void test_ipv6_without_port(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, kCases[i], "80", + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, kCases[i], "80", args.pollset_set, grpc_closure_create(must_succeed, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -138,7 +192,8 @@ static void test_invalid_ip_addresses(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, kCases[i], NULL, + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, kCases[i], NULL, args.pollset_set, grpc_closure_create(must_fail, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args); @@ -154,7 +209,8 @@ static void test_unparseable_hostports(void) { args_struct args; args_init(&args); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_resolve_address(&exec_ctx, kCases[i], "1", + poll_pollset_until_request_done(&args); + grpc_resolve_address(&exec_ctx, kCases[i], "1", args.pollset_set, grpc_closure_create(must_fail, &args), &args.addrs); grpc_exec_ctx_finish(&exec_ctx); args_finish(&args);