diff options
Diffstat (limited to 'test/core/end2end')
93 files changed, 771 insertions, 513 deletions
diff --git a/test/core/end2end/BUILD b/test/core/end2end/BUILD index 0cef7aa01d..ffea1cc4e8 100644 --- a/test/core/end2end/BUILD +++ b/test/core/end2end/BUILD @@ -32,49 +32,66 @@ licenses(["notice"]) # 3-clause BSD load(":generate_tests.bzl", "grpc_end2end_tests") cc_library( - name = 'cq_verifier', - srcs = ['cq_verifier.c'], - hdrs = ['cq_verifier.h'], - deps = ['//:gpr', '//:grpc', '//test/core/util:grpc_test_util'], - copts = ['-std=c99'], - visibility = ["//test:__subpackages__"], + name = "cq_verifier", + srcs = ["cq_verifier.c"], + hdrs = ["cq_verifier.h"], + copts = ["-std=c99"], + visibility = ["//test:__subpackages__"], + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:grpc_test_util", + ], ) cc_library( - name = 'ssl_test_data', - visibility = ["//test:__subpackages__"], - hdrs = ['data/ssl_test_data.h'], - copts = ['-std=c99'], - srcs = [ - "data/client_certs.c", - "data/server1_cert.c", - "data/server1_key.c", - "data/test_root_cert.c", - ] + name = "ssl_test_data", + srcs = [ + "data/client_certs.c", + "data/server1_cert.c", + "data/server1_key.c", + "data/test_root_cert.c", + ], + hdrs = ["data/ssl_test_data.h"], + copts = ["-std=c99"], + visibility = ["//test:__subpackages__"], ) cc_library( - name = 'fake_resolver', - hdrs = ['fake_resolver.h'], - srcs = ['fake_resolver.c'], - copts = ['-std=c99'], - deps = ['//:gpr', '//:grpc', '//test/core/util:grpc_test_util'] + name = "fake_resolver", + srcs = ["fake_resolver.c"], + hdrs = ["fake_resolver.h"], + copts = ["-std=c99"], + visibility = ["//test:__subpackages__"], + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:grpc_test_util", + ], ) cc_library( - name = 'http_proxy', - hdrs = ['fixtures/http_proxy_fixture.h'], - srcs = ['fixtures/http_proxy_fixture.c'], - copts = ['-std=c99'], - deps = ['//:gpr', '//:grpc', '//test/core/util:grpc_test_util'] + name = "http_proxy", + srcs = ["fixtures/http_proxy_fixture.c"], + hdrs = ["fixtures/http_proxy_fixture.h"], + copts = ["-std=c99"], + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:grpc_test_util", + ], ) cc_library( - name = 'proxy', - hdrs = ['fixtures/proxy.h'], - srcs = ['fixtures/proxy.c'], - copts = ['-std=c99'], - deps = ['//:gpr', '//:grpc', '//test/core/util:grpc_test_util'] + name = "proxy", + srcs = ["fixtures/proxy.c"], + hdrs = ["fixtures/proxy.h"], + copts = ["-std=c99"], + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:grpc_test_util", + ], ) grpc_end2end_tests() diff --git a/test/core/end2end/bad_server_response_test.c b/test/core/end2end/bad_server_response_test.c index c37a292af9..fe7e674d17 100644 --- a/test/core/end2end/bad_server_response_test.c +++ b/test/core/end2end/bad_server_response_test.c @@ -178,7 +178,7 @@ static void start_rpc(int target_port, grpc_status_code expected_status, cq_verifier *cqv; grpc_slice details; - state.cq = grpc_completion_queue_create(NULL); + state.cq = grpc_completion_queue_create_for_next(NULL); cqv = cq_verifier_create(state.cq); gpr_join_host_port(&state.target, "127.0.0.1", target_port); state.channel = grpc_insecure_channel_create(state.target, NULL, NULL); @@ -236,7 +236,7 @@ static void cleanup_rpc(grpc_exec_ctx *exec_ctx) { grpc_event ev; grpc_slice_buffer_destroy_internal(exec_ctx, &state.temp_incoming_buffer); grpc_slice_buffer_destroy_internal(exec_ctx, &state.outgoing_buffer); - grpc_call_destroy(state.call); + grpc_call_unref(state.call); grpc_completion_queue_shutdown(state.cq); do { ev = grpc_completion_queue_next(state.cq, n_sec_deadline(1), NULL); diff --git a/test/core/end2end/connection_refused_test.c b/test/core/end2end/connection_refused_test.c index 6ded12ad48..db7a6c0a8e 100644 --- a/test/core/end2end/connection_refused_test.c +++ b/test/core/end2end/connection_refused_test.c @@ -68,7 +68,7 @@ static void run_test(bool wait_for_ready, bool use_service_config) { grpc_metadata_array_init(&trailing_metadata_recv); - cq = grpc_completion_queue_create(NULL); + cq = grpc_completion_queue_create_for_next(NULL); cqv = cq_verifier_create(cq); /* if using service config, create channel args */ @@ -138,7 +138,7 @@ static void run_test(bool wait_for_ready, bool use_service_config) { .type != GRPC_QUEUE_SHUTDOWN) ; grpc_completion_queue_destroy(cq); - grpc_call_destroy(call); + grpc_call_unref(call); grpc_channel_destroy(chan); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index 3623bd7be8..caf7828691 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -76,6 +76,7 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_channel *client; grpc_server *server; grpc_completion_queue *cq; + grpc_completion_queue *shutdown_cq; grpc_call *c; grpc_call *s; cq_verifier *cqv; @@ -107,7 +108,7 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_call_details_init(&call_details); /* Create server. */ - cq = grpc_completion_queue_create(NULL); + cq = grpc_completion_queue_create_for_next(NULL); server = grpc_server_create(NULL, NULL); grpc_server_register_completion_queue(server, cq, NULL); GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port( @@ -242,7 +243,7 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_slice_str_cmp(call_details.host, "foo.test.google.fr")); GPR_ASSERT(was_cancelled == 1); - grpc_call_destroy(s); + grpc_call_unref(s); } else { /* Check for a failed connection. */ CQ_EXPECT_COMPLETION(cqv, tag(1), 1); @@ -251,7 +252,7 @@ void test_connect(const char *server_host, const char *client_host, int port, GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE); } - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); @@ -259,11 +260,14 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_channel_destroy(client); /* Destroy server. */ - grpc_server_shutdown_and_notify(server, cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); + grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(server); + grpc_completion_queue_destroy(shutdown_cq); grpc_completion_queue_shutdown(cq); drain_cq(cq); grpc_completion_queue_destroy(cq); diff --git a/test/core/end2end/end2end_tests.h b/test/core/end2end/end2end_tests.h index cdb26a67e9..4d98bddbd8 100644 --- a/test/core/end2end/end2end_tests.h +++ b/test/core/end2end/end2end_tests.h @@ -53,6 +53,7 @@ typedef struct grpc_end2end_test_config grpc_end2end_test_config; struct grpc_end2end_test_fixture { grpc_completion_queue *cq; + grpc_completion_queue *shutdown_cq; grpc_server *server; grpc_channel *client; void *fixture_data; diff --git a/test/core/end2end/fake_resolver.c b/test/core/end2end/fake_resolver.c index 1c7dd1339c..6a71c20b80 100644 --- a/test/core/end2end/fake_resolver.c +++ b/test/core/end2end/fake_resolver.c @@ -32,6 +32,7 @@ // This is similar to the sockaddr resolver, except that it supports a // bunch of query args that are useful for dependency injection in tests. +#include <limits.h> #include <stdbool.h> #include <stdio.h> #include <stdlib.h> @@ -46,12 +47,18 @@ #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" #include "src/core/lib/channel/channel_args.h" +#include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/lib/iomgr/unix_sockets_posix.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" +#include "test/core/end2end/fake_resolver.h" + +#define GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR \ + "grpc.fake_resolver.response_generator" + // // fake_resolver // @@ -62,12 +69,11 @@ typedef struct { // passed-in parameters grpc_channel_args* channel_args; - grpc_lb_addresses* addresses; - // mutex guarding the rest of the state - gpr_mu mu; - // have we published? - bool published; + // If not NULL, the next set of resolution results to be returned to + // grpc_resolver_next_locked()'s closure. + grpc_channel_args* next_results; + // pending next completion, or NULL grpc_closure* next_completion; // target result address for next completion @@ -76,60 +82,137 @@ typedef struct { static void fake_resolver_destroy(grpc_exec_ctx* exec_ctx, grpc_resolver* gr) { fake_resolver* r = (fake_resolver*)gr; - gpr_mu_destroy(&r->mu); + grpc_channel_args_destroy(exec_ctx, r->next_results); grpc_channel_args_destroy(exec_ctx, r->channel_args); - grpc_lb_addresses_destroy(exec_ctx, r->addresses); gpr_free(r); } -static void fake_resolver_shutdown(grpc_exec_ctx* exec_ctx, - grpc_resolver* resolver) { +static void fake_resolver_shutdown_locked(grpc_exec_ctx* exec_ctx, + grpc_resolver* resolver) { fake_resolver* r = (fake_resolver*)resolver; - gpr_mu_lock(&r->mu); if (r->next_completion != NULL) { *r->target_result = NULL; grpc_closure_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE); r->next_completion = NULL; } - gpr_mu_unlock(&r->mu); } static void fake_resolver_maybe_finish_next_locked(grpc_exec_ctx* exec_ctx, fake_resolver* r) { - if (r->next_completion != NULL && !r->published) { - r->published = true; - grpc_arg arg = grpc_lb_addresses_create_channel_arg(r->addresses); + if (r->next_completion != NULL && r->next_results != NULL) { *r->target_result = - grpc_channel_args_copy_and_add(r->channel_args, &arg, 1); + grpc_channel_args_merge(r->channel_args, r->next_results); + grpc_channel_args_destroy(exec_ctx, r->next_results); grpc_closure_sched(exec_ctx, r->next_completion, GRPC_ERROR_NONE); r->next_completion = NULL; + r->next_results = NULL; } } -static void fake_resolver_channel_saw_error(grpc_exec_ctx* exec_ctx, - grpc_resolver* resolver) { +static void fake_resolver_channel_saw_error_locked(grpc_exec_ctx* exec_ctx, + grpc_resolver* resolver) { fake_resolver* r = (fake_resolver*)resolver; - gpr_mu_lock(&r->mu); - r->published = false; fake_resolver_maybe_finish_next_locked(exec_ctx, r); - gpr_mu_unlock(&r->mu); } -static void fake_resolver_next(grpc_exec_ctx* exec_ctx, grpc_resolver* resolver, - grpc_channel_args** target_result, - grpc_closure* on_complete) { +static void fake_resolver_next_locked(grpc_exec_ctx* exec_ctx, + grpc_resolver* resolver, + grpc_channel_args** target_result, + grpc_closure* on_complete) { fake_resolver* r = (fake_resolver*)resolver; - gpr_mu_lock(&r->mu); GPR_ASSERT(!r->next_completion); r->next_completion = on_complete; r->target_result = target_result; fake_resolver_maybe_finish_next_locked(exec_ctx, r); - gpr_mu_unlock(&r->mu); } static const grpc_resolver_vtable fake_resolver_vtable = { - fake_resolver_destroy, fake_resolver_shutdown, - fake_resolver_channel_saw_error, fake_resolver_next}; + fake_resolver_destroy, fake_resolver_shutdown_locked, + fake_resolver_channel_saw_error_locked, fake_resolver_next_locked}; + +struct grpc_fake_resolver_response_generator { + fake_resolver* resolver; // Set by the fake_resolver constructor to itself. + grpc_channel_args* next_response; + gpr_refcount refcount; +}; + +grpc_fake_resolver_response_generator* +grpc_fake_resolver_response_generator_create() { + grpc_fake_resolver_response_generator* generator = + gpr_zalloc(sizeof(*generator)); + gpr_ref_init(&generator->refcount, 1); + return generator; +} + +grpc_fake_resolver_response_generator* +grpc_fake_resolver_response_generator_ref( + grpc_fake_resolver_response_generator* generator) { + gpr_ref(&generator->refcount); + return generator; +} + +void grpc_fake_resolver_response_generator_unref( + grpc_fake_resolver_response_generator* generator) { + if (gpr_unref(&generator->refcount)) { + gpr_free(generator); + } +} + +static void set_response_cb(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + grpc_fake_resolver_response_generator* generator = arg; + fake_resolver* r = generator->resolver; + if (r->next_results != NULL) { + grpc_channel_args_destroy(exec_ctx, r->next_results); + } + r->next_results = generator->next_response; + fake_resolver_maybe_finish_next_locked(exec_ctx, r); +} + +void grpc_fake_resolver_response_generator_set_response( + grpc_exec_ctx* exec_ctx, grpc_fake_resolver_response_generator* generator, + grpc_channel_args* next_response) { + GPR_ASSERT(generator->resolver != NULL); + generator->next_response = grpc_channel_args_copy(next_response); + grpc_closure_sched( + exec_ctx, + grpc_closure_create( + set_response_cb, generator, + grpc_combiner_scheduler(generator->resolver->base.combiner, false)), + GRPC_ERROR_NONE); +} + +static void* response_generator_arg_copy(void* p) { + return grpc_fake_resolver_response_generator_ref(p); +} + +static void response_generator_arg_destroy(grpc_exec_ctx* exec_ctx, void* p) { + grpc_fake_resolver_response_generator_unref(p); +} + +static int response_generator_cmp(void* a, void* b) { return GPR_ICMP(a, b); } + +static const grpc_arg_pointer_vtable response_generator_arg_vtable = { + response_generator_arg_copy, response_generator_arg_destroy, + response_generator_cmp}; + +grpc_arg grpc_fake_resolver_response_generator_arg( + grpc_fake_resolver_response_generator* generator) { + grpc_arg arg; + arg.type = GRPC_ARG_POINTER; + arg.key = GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR; + arg.value.pointer.p = generator; + arg.value.pointer.vtable = &response_generator_arg_vtable; + return arg; +} + +grpc_fake_resolver_response_generator* +grpc_fake_resolver_get_response_generator(const grpc_channel_args* args) { + const grpc_arg* arg = + grpc_channel_args_find(args, GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR); + if (arg == NULL || arg->type != GRPC_ARG_POINTER) return NULL; + return arg->value.pointer.p; +} // // fake_resolver_factory @@ -139,81 +222,15 @@ static void fake_resolver_factory_ref(grpc_resolver_factory* factory) {} static void fake_resolver_factory_unref(grpc_resolver_factory* factory) {} -static void do_nothing(void* ignored) {} - static grpc_resolver* fake_resolver_create(grpc_exec_ctx* exec_ctx, grpc_resolver_factory* factory, grpc_resolver_args* args) { - if (0 != strcmp(args->uri->authority, "")) { - gpr_log(GPR_ERROR, "authority based uri's not supported by the %s scheme", - args->uri->scheme); - return NULL; - } - // Get lb_enabled arg. Anything other than "0" is interpreted as true. - const char* lb_enabled_qpart = - grpc_uri_get_query_arg(args->uri, "lb_enabled"); - const bool lb_enabled = - lb_enabled_qpart != NULL && strcmp("0", lb_enabled_qpart) != 0; - - // Get the balancer's names. - const char* balancer_names = - grpc_uri_get_query_arg(args->uri, "balancer_names"); - grpc_slice_buffer balancer_names_parts; - grpc_slice_buffer_init(&balancer_names_parts); - if (balancer_names != NULL) { - const grpc_slice balancer_names_slice = - grpc_slice_from_copied_string(balancer_names); - grpc_slice_split(balancer_names_slice, ",", &balancer_names_parts); - grpc_slice_unref(balancer_names_slice); - } - - // Construct addresses. - grpc_slice path_slice = - grpc_slice_new(args->uri->path, strlen(args->uri->path), do_nothing); - grpc_slice_buffer path_parts; - grpc_slice_buffer_init(&path_parts); - grpc_slice_split(path_slice, ",", &path_parts); - if (balancer_names_parts.count > 0 && - path_parts.count != balancer_names_parts.count) { - gpr_log(GPR_ERROR, - "Balancer names present but mismatched with number of addresses: " - "%lu balancer names != %lu addresses", - (unsigned long)balancer_names_parts.count, - (unsigned long)path_parts.count); - return NULL; - } - grpc_lb_addresses* addresses = - grpc_lb_addresses_create(path_parts.count, NULL /* user_data_vtable */); - bool errors_found = false; - for (size_t i = 0; i < addresses->num_addresses; i++) { - grpc_uri ith_uri = *args->uri; - char* part_str = grpc_slice_to_c_string(path_parts.slices[i]); - ith_uri.path = part_str; - if (!parse_ipv4(&ith_uri, &addresses->addresses[i].address)) { - errors_found = true; - } - gpr_free(part_str); - if (errors_found) break; - addresses->addresses[i].is_balancer = lb_enabled; - addresses->addresses[i].balancer_name = - balancer_names_parts.count > 0 - ? grpc_dump_slice(balancer_names_parts.slices[i], GPR_DUMP_ASCII) - : NULL; - } - grpc_slice_buffer_destroy_internal(exec_ctx, &path_parts); - grpc_slice_buffer_destroy_internal(exec_ctx, &balancer_names_parts); - grpc_slice_unref(path_slice); - if (errors_found) { - grpc_lb_addresses_destroy(exec_ctx, addresses); - return NULL; - } - // Instantiate resolver. - fake_resolver* r = gpr_malloc(sizeof(fake_resolver)); - memset(r, 0, sizeof(*r)); + fake_resolver* r = gpr_zalloc(sizeof(*r)); r->channel_args = grpc_channel_args_copy(args->args); - r->addresses = addresses; - gpr_mu_init(&r->mu); grpc_resolver_init(&r->base, &fake_resolver_vtable, args->combiner); + grpc_fake_resolver_response_generator* response_generator = + grpc_fake_resolver_get_response_generator(args->args); + if (response_generator != NULL) response_generator->resolver = r; return &r->base; } diff --git a/test/core/end2end/fake_resolver.h b/test/core/end2end/fake_resolver.h index 7a30347f30..447289adef 100644 --- a/test/core/end2end/fake_resolver.h +++ b/test/core/end2end/fake_resolver.h @@ -32,8 +32,42 @@ #ifndef GRPC_TEST_CORE_END2END_FAKE_RESOLVER_H #define GRPC_TEST_CORE_END2END_FAKE_RESOLVER_H +#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/ext/filters/client_channel/uri_parser.h" +#include "src/core/lib/channel/channel_args.h" + #include "test/core/util/test_config.h" void grpc_fake_resolver_init(); +// Instances of \a grpc_fake_resolver_response_generator are passed to the +// fake resolver in a channel argument (see \a +// grpc_fake_resolver_response_generator_arg) in order to inject and trigger +// custom resolutions. See also \a +// grpc_fake_resolver_response_generator_set_response. +typedef struct grpc_fake_resolver_response_generator + grpc_fake_resolver_response_generator; +grpc_fake_resolver_response_generator* +grpc_fake_resolver_response_generator_create(); + +// Instruct the fake resolver associated with the \a response_generator instance +// to trigger a new resolution for \a uri and \a args. +void grpc_fake_resolver_response_generator_set_response( + grpc_exec_ctx* exec_ctx, grpc_fake_resolver_response_generator* generator, + grpc_channel_args* next_response); + +// Return a \a grpc_arg for a \a grpc_fake_resolver_response_generator instance. +grpc_arg grpc_fake_resolver_response_generator_arg( + grpc_fake_resolver_response_generator* generator); +// Return the \a grpc_fake_resolver_response_generator instance in \a args or +// NULL. +grpc_fake_resolver_response_generator* +grpc_fake_resolver_get_response_generator(const grpc_channel_args* args); + +grpc_fake_resolver_response_generator* +grpc_fake_resolver_response_generator_ref( + grpc_fake_resolver_response_generator* generator); +void grpc_fake_resolver_response_generator_unref( + grpc_fake_resolver_response_generator* generator); + #endif /* GRPC_TEST_CORE_END2END_FAKE_RESOLVER_H */ diff --git a/test/core/end2end/fixtures/h2_census.c b/test/core/end2end/fixtures/h2_census.c index 97b27b2496..e8af03a52b 100644 --- a/test/core/end2end/fixtures/h2_census.c +++ b/test/core/end2end/fixtures/h2_census.c @@ -42,10 +42,10 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" #include "test/core/util/port.h" @@ -65,7 +65,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_compress.c b/test/core/end2end/fixtures/h2_compress.c index 8aec94d601..49fcd9e9d3 100644 --- a/test/core/end2end/fixtures/h2_compress.c +++ b/test/core/end2end/fixtures/h2_compress.c @@ -42,10 +42,10 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" #include "test/core/util/port.h" @@ -69,7 +69,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack_compression( memset(&f, 0, sizeof(f)); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_fakesec.c b/test/core/end2end/fixtures/h2_fakesec.c index c9747913c2..5969b110e6 100644 --- a/test/core/end2end/fixtures/h2_fakesec.c +++ b/test/core/end2end/fixtures/h2_fakesec.c @@ -60,7 +60,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_fd.c b/test/core/end2end/fixtures/h2_fd.c index 223fadc386..53888dbc5b 100644 --- a/test/core/end2end/fixtures/h2_fd.c +++ b/test/core/end2end/fixtures/h2_fd.c @@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = fixture_data; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); create_sockets(fixture_data->fd_pair); diff --git a/test/core/end2end/fixtures/h2_full+pipe.c b/test/core/end2end/fixtures/h2_full+pipe.c index 0191e59fc8..cee48af66f 100644 --- a/test/core/end2end/fixtures/h2_full+pipe.c +++ b/test/core/end2end/fixtures/h2_full+pipe.c @@ -47,9 +47,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/iomgr/wakeup_fd_posix.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" @@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_full+trace.c b/test/core/end2end/fixtures/h2_full+trace.c index 9dbb27fc4b..57047bc57b 100644 --- a/test/core/end2end/fixtures/h2_full+trace.c +++ b/test/core/end2end/fixtures/h2_full+trace.c @@ -47,9 +47,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/support/env.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" @@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_full.c b/test/core/end2end/fixtures/h2_full.c index 49c62b3429..f18d314c1f 100644 --- a/test/core/end2end/fixtures/h2_full.c +++ b/test/core/end2end/fixtures/h2_full.c @@ -42,9 +42,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" #include "test/core/util/port.h" @@ -64,7 +64,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_http_proxy.c b/test/core/end2end/fixtures/h2_http_proxy.c index 62c435557d..aec874b1fa 100644 --- a/test/core/end2end/fixtures/h2_http_proxy.c +++ b/test/core/end2end/fixtures/h2_http_proxy.c @@ -43,9 +43,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/support/env.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" @@ -69,7 +69,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( ffd->proxy = grpc_end2end_http_proxy_create(); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_load_reporting.c b/test/core/end2end/fixtures/h2_load_reporting.c index 79f26ed2bc..8e9c885759 100644 --- a/test/core/end2end/fixtures/h2_load_reporting.c +++ b/test/core/end2end/fixtures/h2_load_reporting.c @@ -42,11 +42,11 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/filters/load_reporting/load_reporting.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" #include "test/core/util/port.h" @@ -67,7 +67,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_load_reporting( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_oauth2.c b/test/core/end2end/fixtures/h2_oauth2.c index 3351652858..c94f1f6239 100644 --- a/test/core/end2end/fixtures/h2_oauth2.c +++ b/test/core/end2end/fixtures/h2_oauth2.c @@ -113,7 +113,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_proxy.c b/test/core/end2end/fixtures/h2_proxy.c index a10738fa0b..3d33d5860d 100644 --- a/test/core/end2end/fixtures/h2_proxy.c +++ b/test/core/end2end/fixtures/h2_proxy.c @@ -42,9 +42,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" #include "test/core/end2end/fixtures/proxy.h" @@ -79,7 +79,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_sockpair+trace.c b/test/core/end2end/fixtures/h2_sockpair+trace.c index 424241c1e4..f3e12200f8 100644 --- a/test/core/end2end/fixtures/h2_sockpair+trace.c +++ b/test/core/end2end/fixtures/h2_sockpair+trace.c @@ -46,11 +46,11 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/client/http_client_filter.h" +#include "src/core/ext/filters/http/message_compress/message_compress_filter.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" -#include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_client_filter.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/iomgr/endpoint_pair.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/support/env.h" @@ -94,7 +94,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); *sfd = grpc_iomgr_create_endpoint_pair("fixture", NULL); diff --git a/test/core/end2end/fixtures/h2_sockpair.c b/test/core/end2end/fixtures/h2_sockpair.c index fe8d766e74..4e60d03a9c 100644 --- a/test/core/end2end/fixtures/h2_sockpair.c +++ b/test/core/end2end/fixtures/h2_sockpair.c @@ -41,11 +41,11 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/client/http_client_filter.h" +#include "src/core/ext/filters/http/message_compress/message_compress_filter.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" -#include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_client_filter.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/iomgr/endpoint_pair.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/surface/channel.h" @@ -88,7 +88,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); *sfd = grpc_iomgr_create_endpoint_pair("fixture", NULL); diff --git a/test/core/end2end/fixtures/h2_sockpair_1byte.c b/test/core/end2end/fixtures/h2_sockpair_1byte.c index 04174fa501..8714266655 100644 --- a/test/core/end2end/fixtures/h2_sockpair_1byte.c +++ b/test/core/end2end/fixtures/h2_sockpair_1byte.c @@ -41,11 +41,11 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/client/http_client_filter.h" +#include "src/core/ext/filters/http/message_compress/message_compress_filter.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" -#include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_client_filter.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/iomgr/endpoint_pair.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/surface/channel.h" @@ -88,7 +88,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); grpc_arg a[] = {{.key = GRPC_ARG_TCP_READ_CHUNK_SIZE, .type = GRPC_ARG_INTEGER, diff --git a/test/core/end2end/fixtures/h2_ssl.c b/test/core/end2end/fixtures/h2_ssl.c index cf44cd093c..c6a1ca09f8 100644 --- a/test/core/end2end/fixtures/h2_ssl.c +++ b/test/core/end2end/fixtures/h2_ssl.c @@ -64,7 +64,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_ssl_cert.c b/test/core/end2end/fixtures/h2_ssl_cert.c index f62331eea3..ba3956e8e3 100644 --- a/test/core/end2end/fixtures/h2_ssl_cert.c +++ b/test/core/end2end/fixtures/h2_ssl_cert.c @@ -67,7 +67,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } @@ -289,9 +290,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -310,6 +312,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_fixture f, @@ -340,7 +343,7 @@ static void simple_request_body(grpc_end2end_test_fixture f, CQ_EXPECT_COMPLETION(cqv, tag(1), expected_result == SUCCESS); cq_verify(cqv); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/fixtures/h2_ssl_proxy.c b/test/core/end2end/fixtures/h2_ssl_proxy.c index 740b075bf6..9a6c9f558f 100644 --- a/test/core/end2end/fixtures/h2_ssl_proxy.c +++ b/test/core/end2end/fixtures/h2_ssl_proxy.c @@ -100,7 +100,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/h2_uds.c b/test/core/end2end/fixtures/h2_uds.c index 7bde69d82a..de1b8eb5df 100644 --- a/test/core/end2end/fixtures/h2_uds.c +++ b/test/core/end2end/fixtures/h2_uds.c @@ -45,9 +45,9 @@ #include <grpc/support/thd.h> #include <grpc/support/useful.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/http/server/http_server_filter.h" #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/connected_channel.h" -#include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/server.h" @@ -70,7 +70,8 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( unique++); f.fixture_data = ffd; - f.cq = grpc_completion_queue_create(NULL); + f.cq = grpc_completion_queue_create_for_next(NULL); + f.shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); return f; } diff --git a/test/core/end2end/fixtures/http_proxy_fixture.c b/test/core/end2end/fixtures/http_proxy_fixture.c index 451ed268d3..f0d09487c6 100644 --- a/test/core/end2end/fixtures/http_proxy_fixture.c +++ b/test/core/end2end/fixtures/http_proxy_fixture.c @@ -59,6 +59,7 @@ #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/iomgr/tcp_client.h" #include "src/core/lib/iomgr/tcp_server.h" +#include "src/core/lib/iomgr/timer.h" #include "src/core/lib/slice/slice_internal.h" #include "test/core/util/port.h" @@ -69,7 +70,7 @@ struct grpc_end2end_http_proxy { grpc_channel_args* channel_args; gpr_mu* mu; grpc_pollset* pollset; - gpr_atm shutdown; + gpr_refcount users; }; // @@ -77,6 +78,8 @@ struct grpc_end2end_http_proxy { // typedef struct proxy_connection { + grpc_end2end_http_proxy* proxy; + grpc_endpoint* client_endpoint; grpc_endpoint* server_endpoint; @@ -103,13 +106,20 @@ typedef struct proxy_connection { grpc_http_request http_request; } proxy_connection; +static void proxy_connection_ref(proxy_connection* conn, const char* reason) { + gpr_ref(&conn->refcount); +} + // Helper function to destroy the proxy connection. static void proxy_connection_unref(grpc_exec_ctx* exec_ctx, - proxy_connection* conn) { + proxy_connection* conn, const char* reason) { if (gpr_unref(&conn->refcount)) { + gpr_log(GPR_DEBUG, "endpoints: %p %p", conn->client_endpoint, + conn->server_endpoint); grpc_endpoint_destroy(exec_ctx, conn->client_endpoint); - if (conn->server_endpoint != NULL) + if (conn->server_endpoint != NULL) { grpc_endpoint_destroy(exec_ctx, conn->server_endpoint); + } grpc_pollset_set_destroy(exec_ctx, conn->pollset_set); grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_read_buffer); grpc_slice_buffer_destroy_internal(exec_ctx, @@ -121,6 +131,7 @@ static void proxy_connection_unref(grpc_exec_ctx* exec_ctx, grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_write_buffer); grpc_http_parser_destroy(&conn->http_parser); grpc_http_request_destroy(&conn->http_request); + gpr_unref(&conn->proxy->users); gpr_free(conn); } } @@ -139,7 +150,7 @@ static void proxy_connection_failed(grpc_exec_ctx* exec_ctx, grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint, GRPC_ERROR_REF(error)); } - proxy_connection_unref(exec_ctx, conn); + proxy_connection_unref(exec_ctx, conn, "conn_failed"); } // Callback for writing proxy data to the client. @@ -163,7 +174,7 @@ static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg, &conn->on_client_write_done); } else { // No more writes. Unref the connection. - proxy_connection_unref(exec_ctx, conn); + proxy_connection_unref(exec_ctx, conn, "write_done"); } } @@ -188,7 +199,7 @@ static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg, &conn->on_server_write_done); } else { // No more writes. Unref the connection. - proxy_connection_unref(exec_ctx, conn); + proxy_connection_unref(exec_ctx, conn, "server_write"); } } @@ -214,7 +225,7 @@ static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg, } else { grpc_slice_buffer_move_into(&conn->client_read_buffer, &conn->server_write_buffer); - gpr_ref(&conn->refcount); + proxy_connection_ref(conn, "client_read"); grpc_endpoint_write(exec_ctx, conn->server_endpoint, &conn->server_write_buffer, &conn->on_server_write_done); @@ -246,7 +257,7 @@ static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg, } else { grpc_slice_buffer_move_into(&conn->server_read_buffer, &conn->client_write_buffer); - gpr_ref(&conn->refcount); + proxy_connection_ref(conn, "server_read"); grpc_endpoint_write(exec_ctx, conn->client_endpoint, &conn->client_write_buffer, &conn->on_client_write_done); @@ -270,7 +281,9 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg, // Start reading from both client and server. One of the read // requests inherits our ref to conn, but we need to take a new ref // for the other one. - gpr_ref(&conn->refcount); + proxy_connection_ref(conn, "client_read"); + proxy_connection_ref(conn, "server_read"); + proxy_connection_unref(exec_ctx, conn, "write_response"); grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer, &conn->on_client_read_done); grpc_endpoint_read(exec_ctx, conn->server_endpoint, &conn->server_read_buffer, @@ -312,6 +325,8 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg, static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { proxy_connection* conn = arg; + gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn, + grpc_error_string(error)); if (error != GRPC_ERROR_NONE) { proxy_connection_failed(exec_ctx, conn, true /* is_client */, "HTTP proxy read request", error); @@ -376,12 +391,14 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, gpr_free(acceptor); grpc_end2end_http_proxy* proxy = arg; // Instantiate proxy_connection. - proxy_connection* conn = gpr_malloc(sizeof(*conn)); - memset(conn, 0, sizeof(*conn)); + proxy_connection* conn = gpr_zalloc(sizeof(*conn)); + gpr_ref(&proxy->users); conn->client_endpoint = endpoint; + conn->proxy = proxy; gpr_ref_init(&conn->refcount, 1); conn->pollset_set = grpc_pollset_set_create(); grpc_pollset_set_add_pollset(exec_ctx, conn->pollset_set, proxy->pollset); + grpc_endpoint_add_to_pollset_set(exec_ctx, endpoint, conn->pollset_set); grpc_closure_init(&conn->on_read_request_done, on_read_request_done, conn, grpc_schedule_on_exec_ctx); grpc_closure_init(&conn->on_server_connect_done, on_server_connect_done, conn, @@ -416,6 +433,7 @@ static void thread_main(void* arg) { grpc_end2end_http_proxy* proxy = arg; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; do { + gpr_ref(&proxy->users); const gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); const gpr_timespec deadline = gpr_time_add(now, gpr_time_from_seconds(1, GPR_TIMESPAN)); @@ -426,7 +444,7 @@ static void thread_main(void* arg) { grpc_pollset_work(&exec_ctx, proxy->pollset, &worker, now, deadline)); gpr_mu_unlock(proxy->mu); grpc_exec_ctx_flush(&exec_ctx); - } while (!gpr_atm_acq_load(&proxy->shutdown)); + } while (!gpr_unref(&proxy->users)); grpc_exec_ctx_finish(&exec_ctx); } @@ -434,6 +452,7 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_end2end_http_proxy* proxy = gpr_malloc(sizeof(*proxy)); memset(proxy, 0, sizeof(*proxy)); + gpr_ref_init(&proxy->users, 1); // Construct proxy address. const int proxy_port = grpc_pick_unused_port_or_die(); gpr_join_host_port(&proxy->proxy_name, "localhost", proxy_port); @@ -474,17 +493,16 @@ static void destroy_pollset(grpc_exec_ctx* exec_ctx, void* arg, } void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) { - gpr_atm_rel_store(&proxy->shutdown, 1); // Signal proxy thread to shutdown. + gpr_unref(&proxy->users); // Signal proxy thread to shutdown. grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; gpr_thd_join(proxy->thd); grpc_tcp_server_shutdown_listeners(&exec_ctx, proxy->server); grpc_tcp_server_unref(&exec_ctx, proxy->server); gpr_free(proxy->proxy_name); grpc_channel_args_destroy(&exec_ctx, proxy->channel_args); - grpc_closure destroyed; - grpc_closure_init(&destroyed, destroy_pollset, proxy->pollset, - grpc_schedule_on_exec_ctx); - grpc_pollset_shutdown(&exec_ctx, proxy->pollset, &destroyed); + grpc_pollset_shutdown(&exec_ctx, proxy->pollset, + grpc_closure_create(destroy_pollset, proxy->pollset, + grpc_schedule_on_exec_ctx)); gpr_free(proxy); grpc_exec_ctx_finish(&exec_ctx); } diff --git a/test/core/end2end/fixtures/proxy.c b/test/core/end2end/fixtures/proxy.c index cee053e8c5..d660073ce6 100644 --- a/test/core/end2end/fixtures/proxy.c +++ b/test/core/end2end/fixtures/proxy.c @@ -104,7 +104,7 @@ grpc_end2end_proxy *grpc_end2end_proxy_create(const grpc_end2end_proxy_def *def, gpr_log(GPR_DEBUG, "PROXY ADDR:%s BACKEND:%s", proxy->proxy_port, proxy->server_port); - proxy->cq = grpc_completion_queue_create(NULL); + proxy->cq = grpc_completion_queue_create_for_next(NULL); proxy->server = def->create_server(proxy->proxy_port, server_args); proxy->client = def->create_client(proxy->server_port, client_args); @@ -148,8 +148,8 @@ void grpc_end2end_proxy_destroy(grpc_end2end_proxy *proxy) { static void unrefpc(proxy_call *pc, const char *reason) { if (gpr_unref(&pc->refs)) { - grpc_call_destroy(pc->c2p); - grpc_call_destroy(pc->p2s); + grpc_call_unref(pc->c2p); + grpc_call_unref(pc->p2s); grpc_metadata_array_destroy(&pc->c2p_initial_metadata); grpc_metadata_array_destroy(&pc->p2s_initial_metadata); grpc_metadata_array_destroy(&pc->p2s_trailing_metadata); diff --git a/test/core/end2end/fuzzers/api_fuzzer.c b/test/core/end2end/fuzzers/api_fuzzer.c index a0acf5bf60..88a0e301da 100644 --- a/test/core/end2end/fuzzers/api_fuzzer.c +++ b/test/core/end2end/fuzzers/api_fuzzer.c @@ -661,7 +661,7 @@ static void read_metadata(input_stream *inp, size_t *count, } static call_state *destroy_call(call_state *call) { - grpc_call_destroy(call->call); + grpc_call_unref(call->call); call->call = NULL; return maybe_delete_call_state(call); } @@ -735,7 +735,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { g_active_call = new_call(NULL, ROOT); g_resource_quota = grpc_resource_quota_create("api_fuzzer"); - grpc_completion_queue *cq = grpc_completion_queue_create(NULL); + grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL); while (!is_eof(&inp) || g_channel != NULL || g_server != NULL || pending_channel_watches > 0 || pending_pings > 0 || @@ -932,6 +932,9 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { } uint32_t propagation_mask = read_uint32(&inp); grpc_slice method = read_string_like_slice(&inp); + if (GRPC_SLICE_LENGTH(method) == 0) { + ok = false; + } grpc_slice host = read_string_like_slice(&inp); gpr_timespec deadline = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), @@ -967,7 +970,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { break; } grpc_op *ops = gpr_malloc(sizeof(grpc_op) * num_ops); - memset(ops, 0, sizeof(grpc_op) * num_ops); + if (num_ops > 0) memset(ops, 0, sizeof(grpc_op) * num_ops); bool ok = true; size_t i; grpc_op *op; diff --git a/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-5867145026076672 b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-5867145026076672 Binary files differnew file mode 100644 index 0000000000..3fd5427b46 --- /dev/null +++ b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-5867145026076672 diff --git a/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6462055064272896 b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6462055064272896 Binary files differnew file mode 100644 index 0000000000..c121283242 --- /dev/null +++ b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-6462055064272896 diff --git a/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-minimized-5175380371570688 b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-minimized-5175380371570688 Binary files differnew file mode 100644 index 0000000000..9c7aebc63a --- /dev/null +++ b/test/core/end2end/fuzzers/api_fuzzer_corpus/clusterfuzz-testcase-minimized-5175380371570688 diff --git a/test/core/end2end/fuzzers/client_fuzzer.c b/test/core/end2end/fuzzers/client_fuzzer.c index e7e7dbefd0..6f49baffd2 100644 --- a/test/core/end2end/fuzzers/client_fuzzer.c +++ b/test/core/end2end/fuzzers/client_fuzzer.c @@ -65,7 +65,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_mock_endpoint_create(discard_write, resource_quota); grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); - grpc_completion_queue *cq = grpc_completion_queue_create(NULL); + grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL); grpc_transport *transport = grpc_create_chttp2_transport(&exec_ctx, NULL, mock_endpoint, 1); grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL); @@ -151,7 +151,7 @@ done: ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME), NULL); GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); } - grpc_call_destroy(call); + grpc_call_unref(call); grpc_completion_queue_destroy(cq); grpc_metadata_array_destroy(&initial_metadata_recv); grpc_metadata_array_destroy(&trailing_metadata_recv); diff --git a/test/core/end2end/fuzzers/hpack.dictionary b/test/core/end2end/fuzzers/hpack.dictionary index 6b96785419..2bb9de34c5 100644 --- a/test/core/end2end/fuzzers/hpack.dictionary +++ b/test/core/end2end/fuzzers/hpack.dictionary @@ -10,14 +10,15 @@ "\x10grpc-payload-bin" "\x0Dgrpc-encoding" "\x14grpc-accept-encoding" +"\x15grpc-server-stats-bin" +"\x0Dgrpc-tags-bin" +"\x0Egrpc-trace-bin" "\x0Ccontent-type" "\x1Egrpc-internal-encoding-request" "\x0Auser-agent" "\x04host" "\x08lb-token" "\x0Cgrpc-timeout" -"\x10grpc-tracing-bin" -"\x0Egrpc-stats-bin" "\x00" "\x13grpc.wait_for_ready" "\x0Cgrpc.timeout" @@ -76,6 +77,7 @@ "\x08if-range" "\x13if-unmodified-since" "\x0Dlast-modified" +"\x0Blb-cost-bin" "\x04link" "\x08location" "\x0Cmax-forwards" @@ -152,6 +154,7 @@ "\x00\x13if-unmodified-since\x00" "\x00\x0Dlast-modified\x00" "\x00\x08lb-token\x00" +"\x00\x0Blb-cost-bin\x00" "\x00\x04link\x00" "\x00\x08location\x00" "\x00\x0Cmax-forwards\x00" diff --git a/test/core/end2end/fuzzers/server_fuzzer.c b/test/core/end2end/fuzzers/server_fuzzer.c index 186542d4b2..6d65fe1847 100644 --- a/test/core/end2end/fuzzers/server_fuzzer.c +++ b/test/core/end2end/fuzzers/server_fuzzer.c @@ -67,7 +67,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_slice_from_copied_buffer((const char *)data, size)); grpc_server *server = grpc_server_create(NULL, NULL); - grpc_completion_queue *cq = grpc_completion_queue_create(NULL); + grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL); grpc_server_register_completion_queue(server, cq, NULL); // TODO(ctiller): add registered methods (one for POST, one for PUT) // void *registered_method = @@ -109,7 +109,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { } done: - if (call1 != NULL) grpc_call_destroy(call1); + if (call1 != NULL) grpc_call_unref(call1); grpc_call_details_destroy(&call_details1); grpc_metadata_array_destroy(&request_metadata1); grpc_server_shutdown_and_notify(server, cq, tag(0xdead)); diff --git a/test/core/end2end/fuzzers/server_fuzzer_corpus/clusterfuzz-testcase-5595941564317696 b/test/core/end2end/fuzzers/server_fuzzer_corpus/clusterfuzz-testcase-5595941564317696 Binary files differnew file mode 100644 index 0000000000..335ce87196 --- /dev/null +++ b/test/core/end2end/fuzzers/server_fuzzer_corpus/clusterfuzz-testcase-5595941564317696 diff --git a/test/core/end2end/gen_build_yaml.py b/test/core/end2end/gen_build_yaml.py index d1e510d636..48e5720539 100755 --- a/test/core/end2end/gen_build_yaml.py +++ b/test/core/end2end/gen_build_yaml.py @@ -39,9 +39,9 @@ import hashlib FixtureOptions = collections.namedtuple( 'FixtureOptions', - 'fullstack includes_proxy dns_resolver secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes') + 'fullstack includes_proxy dns_resolver secure platforms ci_mac tracing exclude_configs exclude_iomgrs large_writes enables_compression') default_unsecure_fixture_options = FixtureOptions( - True, False, True, False, ['windows', 'linux', 'mac', 'posix'], True, False, [], [], True) + True, False, True, False, ['windows', 'linux', 'mac', 'posix'], True, False, [], [], True, False) socketpair_unsecure_fixture_options = default_unsecure_fixture_options._replace(fullstack=False, dns_resolver=False) default_secure_fixture_options = default_unsecure_fixture_options._replace(secure=True) uds_fixture_options = default_unsecure_fixture_options._replace(dns_resolver=False, platforms=['linux', 'mac', 'posix'], exclude_iomgrs=['uv']) @@ -51,8 +51,7 @@ fd_unsecure_fixture_options = default_unsecure_fixture_options._replace( # maps fixture name to whether it requires the security library END2END_FIXTURES = { - 'h2_compress': default_unsecure_fixture_options, - + 'h2_compress': default_unsecure_fixture_options._replace(enables_compression=True), 'h2_census': default_unsecure_fixture_options, 'h2_load_reporting': default_unsecure_fixture_options, 'h2_fakesec': default_secure_fixture_options._replace(ci_mac=False), @@ -83,8 +82,8 @@ END2END_FIXTURES = { TestOptions = collections.namedtuple( 'TestOptions', - 'needs_fullstack needs_dns proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky') -default_test_options = TestOptions(False, False, True, False, True, 1.0, [], False, False) + 'needs_fullstack needs_dns proxyable secure traceable cpu_cost exclude_iomgrs large_writes flaky allow_compression') +default_test_options = TestOptions(False, False, True, False, True, 1.0, [], False, False, True) connectivity_test_options = default_test_options._replace(needs_fullstack=True) LOWCPU = 0.1 @@ -94,12 +93,13 @@ END2END_TESTS = { 'authority_not_supported': default_test_options, 'bad_hostname': default_test_options, 'bad_ping': connectivity_test_options._replace(proxyable=False), - 'binary_metadata': default_test_options, + 'binary_metadata': default_test_options._replace(cpu_cost=LOWCPU), 'resource_quota_server': default_test_options._replace(large_writes=True, - proxyable=False), + proxyable=False, + allow_compression=False), 'call_creds': default_test_options._replace(secure=True), 'cancel_after_accept': default_test_options._replace(cpu_cost=LOWCPU), - 'cancel_after_client_done': default_test_options, + 'cancel_after_client_done': default_test_options._replace(cpu_cost=LOWCPU), 'cancel_after_invoke': default_test_options._replace(cpu_cost=LOWCPU), 'cancel_before_invoke': default_test_options._replace(cpu_cost=LOWCPU), 'cancel_in_a_vacuum': default_test_options._replace(cpu_cost=LOWCPU), @@ -110,46 +110,49 @@ END2END_TESTS = { 'default_host': default_test_options._replace(needs_fullstack=True, needs_dns=True), 'disappearing_server': connectivity_test_options._replace(flaky=True), - 'empty_batch': default_test_options, - 'filter_causes_close': default_test_options, + 'empty_batch': default_test_options._replace(cpu_cost=LOWCPU), + 'filter_causes_close': default_test_options._replace(cpu_cost=LOWCPU), 'filter_call_init_fails': default_test_options, - 'filter_latency': default_test_options, + 'filter_latency': default_test_options._replace(cpu_cost=LOWCPU), 'graceful_server_shutdown': default_test_options._replace(cpu_cost=LOWCPU), 'hpack_size': default_test_options._replace(proxyable=False, - traceable=False), - 'high_initial_seqno': default_test_options, + traceable=False, + cpu_cost=LOWCPU), + 'high_initial_seqno': default_test_options._replace(cpu_cost=LOWCPU), 'idempotent_request': default_test_options, 'invoke_large_request': default_test_options, - 'keepalive_timeout': default_test_options._replace(proxyable=False), + 'keepalive_timeout': default_test_options._replace(proxyable=False, + cpu_cost=LOWCPU), 'large_metadata': default_test_options, - 'max_concurrent_streams': default_test_options._replace(proxyable=False), - 'max_connection_age': default_test_options, + 'max_concurrent_streams': default_test_options._replace( + proxyable=False, cpu_cost=LOWCPU), + 'max_connection_age': default_test_options._replace(cpu_cost=LOWCPU), 'max_connection_idle': connectivity_test_options._replace( - proxyable=False, exclude_iomgrs=['uv']), - 'max_message_length': default_test_options, + proxyable=False, exclude_iomgrs=['uv'], cpu_cost=LOWCPU), + 'max_message_length': default_test_options._replace(cpu_cost=LOWCPU), 'negative_deadline': default_test_options, - 'network_status_change': default_test_options, + 'network_status_change': default_test_options._replace(cpu_cost=LOWCPU), 'no_logging': default_test_options._replace(traceable=False), 'no_op': default_test_options, 'payload': default_test_options, 'load_reporting_hook': default_test_options, - 'ping_pong_streaming': default_test_options, - 'ping': connectivity_test_options._replace(proxyable=False), + 'ping_pong_streaming': default_test_options._replace(cpu_cost=LOWCPU), + 'ping': connectivity_test_options._replace(proxyable=False, cpu_cost=LOWCPU), 'registered_call': default_test_options, 'request_with_flags': default_test_options._replace( proxyable=False, cpu_cost=LOWCPU), - 'request_with_payload': default_test_options, - 'server_finishes_request': default_test_options, - 'shutdown_finishes_calls': default_test_options, - 'shutdown_finishes_tags': default_test_options, - 'simple_cacheable_request': default_test_options, + 'request_with_payload': default_test_options._replace(cpu_cost=LOWCPU), + 'server_finishes_request': default_test_options._replace(cpu_cost=LOWCPU), + 'shutdown_finishes_calls': default_test_options._replace(cpu_cost=LOWCPU), + 'shutdown_finishes_tags': default_test_options._replace(cpu_cost=LOWCPU), + 'simple_cacheable_request': default_test_options._replace(cpu_cost=LOWCPU), 'simple_delayed_request': connectivity_test_options, 'simple_metadata': default_test_options, 'simple_request': default_test_options, - 'streaming_error_response': default_test_options, + 'streaming_error_response': default_test_options._replace(cpu_cost=LOWCPU), 'trailing_metadata': default_test_options, - 'write_buffering': default_test_options, - 'write_buffering_at_end': default_test_options, + 'write_buffering': default_test_options._replace(cpu_cost=LOWCPU), + 'write_buffering_at_end': default_test_options._replace(cpu_cost=LOWCPU), } @@ -169,6 +172,9 @@ def compatible(f, t): if END2END_TESTS[t].large_writes: if not END2END_FIXTURES[f].large_writes: return False + if not END2END_TESTS[t].allow_compression: + if END2END_FIXTURES[f].enables_compression: + return False return True diff --git a/test/core/end2end/goaway_server_test.c b/test/core/end2end/goaway_server_test.c index 22d93b321a..ababdb70a8 100644 --- a/test/core/end2end/goaway_server_test.c +++ b/test/core/end2end/goaway_server_test.c @@ -129,7 +129,7 @@ int main(int argc, char **argv) { grpc_metadata_array_init(&request_metadata2); grpc_call_details_init(&request_details2); - cq = grpc_completion_queue_create(NULL); + cq = grpc_completion_queue_create_for_next(NULL); cqv = cq_verifier_create(cq); /* reserve two ports */ @@ -302,10 +302,10 @@ int main(int argc, char **argv) { CQ_EXPECT_COMPLETION(cqv, tag(0xdead2), 1); cq_verify(cqv); - grpc_call_destroy(call1); - grpc_call_destroy(call2); - grpc_call_destroy(server_call1); - grpc_call_destroy(server_call2); + grpc_call_unref(call1); + grpc_call_unref(call2); + grpc_call_unref(server_call1); + grpc_call_unref(server_call2); grpc_server_destroy(server1); grpc_server_destroy(server2); grpc_channel_destroy(chan); diff --git a/test/core/end2end/invalid_call_argument_test.c b/test/core/end2end/invalid_call_argument_test.c index bfd8e6fefa..e9a31f278c 100644 --- a/test/core/end2end/invalid_call_argument_test.c +++ b/test/core/end2end/invalid_call_argument_test.c @@ -75,7 +75,7 @@ static void prepare_test(int is_client) { grpc_metadata_array_init(&g_state.initial_metadata_recv); grpc_metadata_array_init(&g_state.trailing_metadata_recv); g_state.deadline = grpc_timeout_seconds_to_deadline(2); - g_state.cq = grpc_completion_queue_create(NULL); + g_state.cq = grpc_completion_queue_create_for_next(NULL); g_state.cqv = cq_verifier_create(g_state.cq); g_state.details = grpc_empty_slice(); memset(g_state.ops, 0, sizeof(g_state.ops)); @@ -125,7 +125,8 @@ static void prepare_test(int is_client) { } static void cleanup_test() { - grpc_call_destroy(g_state.call); + grpc_completion_queue *shutdown_cq; + grpc_call_unref(g_state.call); cq_verifier_destroy(g_state.cqv); grpc_channel_destroy(g_state.chan); grpc_slice_unref(g_state.details); @@ -133,12 +134,14 @@ static void cleanup_test() { grpc_metadata_array_destroy(&g_state.trailing_metadata_recv); if (!g_state.is_client) { - grpc_call_destroy(g_state.server_call); - grpc_server_shutdown_and_notify(g_state.server, g_state.cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(g_state.cq, tag(1000), + shutdown_cq = grpc_completion_queue_create_for_pluck(NULL); + grpc_call_unref(g_state.server_call); + grpc_server_shutdown_and_notify(g_state.server, shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) .type == GRPC_OP_COMPLETE); + grpc_completion_queue_destroy(shutdown_cq); grpc_server_destroy(g_state.server); grpc_call_details_destroy(&g_state.call_details); grpc_metadata_array_destroy(&g_state.server_initial_metadata_recv); diff --git a/test/core/end2end/multiple_server_queues_test.c b/test/core/end2end/multiple_server_queues_test.c index 5e2eaf4ae9..4696ecd813 100644 --- a/test/core/end2end/multiple_server_queues_test.c +++ b/test/core/end2end/multiple_server_queues_test.c @@ -37,27 +37,50 @@ int main(int argc, char **argv) { grpc_completion_queue *cq1; grpc_completion_queue *cq2; + grpc_completion_queue *cq3; + grpc_completion_queue_attributes attr; + grpc_server *server; grpc_test_init(argc, argv); grpc_init(); - cq1 = grpc_completion_queue_create(NULL); - cq2 = grpc_completion_queue_create(NULL); + + attr.version = 1; + attr.cq_completion_type = GRPC_CQ_NEXT; + attr.cq_polling_type = GRPC_CQ_DEFAULT_POLLING; + cq1 = grpc_completion_queue_create( + grpc_completion_queue_factory_lookup(&attr), &attr, NULL); + + attr.cq_polling_type = GRPC_CQ_NON_LISTENING; + cq2 = grpc_completion_queue_create( + grpc_completion_queue_factory_lookup(&attr), &attr, NULL); + + attr.cq_polling_type = GRPC_CQ_NON_POLLING; + cq3 = grpc_completion_queue_create( + grpc_completion_queue_factory_lookup(&attr), &attr, NULL); + server = grpc_server_create(NULL, NULL); grpc_server_register_completion_queue(server, cq1, NULL); grpc_server_add_insecure_http2_port(server, "[::]:0"); grpc_server_register_completion_queue(server, cq2, NULL); + grpc_server_register_completion_queue(server, cq3, NULL); + grpc_server_start(server); grpc_server_shutdown_and_notify(server, cq2, NULL); grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); /* cue queue hang */ grpc_completion_queue_shutdown(cq1); grpc_completion_queue_shutdown(cq2); + grpc_completion_queue_shutdown(cq3); + grpc_completion_queue_next(cq1, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); + grpc_completion_queue_next(cq3, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); + grpc_server_destroy(server); grpc_completion_queue_destroy(cq1); grpc_completion_queue_destroy(cq2); + grpc_completion_queue_destroy(cq3); grpc_shutdown(); return 0; } diff --git a/test/core/end2end/no_server_test.c b/test/core/end2end/no_server_test.c index 26d26d8f7a..a479d5c868 100644 --- a/test/core/end2end/no_server_test.c +++ b/test/core/end2end/no_server_test.c @@ -59,7 +59,7 @@ int main(int argc, char **argv) { grpc_metadata_array_init(&trailing_metadata_recv); - cq = grpc_completion_queue_create(NULL); + cq = grpc_completion_queue_create_for_next(NULL); cqv = cq_verifier_create(cq); /* create a call, channel to a non existant server */ @@ -97,7 +97,7 @@ int main(int argc, char **argv) { .type != GRPC_QUEUE_SHUTDOWN) ; grpc_completion_queue_destroy(cq); - grpc_call_destroy(call); + grpc_call_unref(call); grpc_channel_destroy(chan); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/authority_not_supported.c b/test/core/end2end/tests/authority_not_supported.c index 192a20d400..13cda08617 100644 --- a/test/core/end2end/tests/authority_not_supported.c +++ b/test/core/end2end/tests/authority_not_supported.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Request/response with metadata and payload.*/ @@ -180,7 +182,7 @@ static void test_with_authority_header(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&initial_metadata_recv); grpc_metadata_array_destroy(&trailing_metadata_recv); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/bad_hostname.c b/test/core/end2end/tests/bad_hostname.c index c63c9c78ad..35a531cf78 100644 --- a/test/core/end2end/tests/bad_hostname.c +++ b/test/core/end2end/tests/bad_hostname.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_fixture f) { @@ -161,7 +163,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/bad_ping.c b/test/core/end2end/tests/bad_ping.c index 01a6aeaa04..0928ba60cf 100644 --- a/test/core/end2end/tests/bad_ping.c +++ b/test/core/end2end/tests/bad_ping.c @@ -75,6 +75,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_bad_ping(grpc_end2end_test_config config) { @@ -207,7 +208,7 @@ static void test_bad_ping(grpc_end2end_test_config config) { CQ_EXPECT_COMPLETION(cqv, tag(0xdead), 1); cq_verify(cqv); - grpc_call_destroy(s); + grpc_call_unref(s); // The connection should be closed immediately after the misbehaved pings, // the in-progress RPC should fail. @@ -223,7 +224,7 @@ static void test_bad_ping(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&trailing_metadata_recv); grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/binary_metadata.c b/test/core/end2end/tests/binary_metadata.c index 9cb17c9d42..3d36849400 100644 --- a/test/core/end2end/tests/binary_metadata.c +++ b/test/core/end2end/tests/binary_metadata.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Request/response with metadata and payload.*/ @@ -312,8 +314,8 @@ static void test_request_response_with_metadata_and_payload( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/call_creds.c b/test/core/end2end/tests/call_creds.c index d65073fbbe..bc430e56b3 100644 --- a/test/core/end2end/tests/call_creds.c +++ b/test/core/end2end/tests/call_creds.c @@ -92,9 +92,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -113,6 +114,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void print_auth_context(int is_client, const grpc_auth_context *ctx) { @@ -345,8 +347,8 @@ static void request_response_with_payload_and_call_creds( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); @@ -471,7 +473,7 @@ static void test_request_with_server_rejecting_client_creds( grpc_byte_buffer_destroy(response_payload_recv); grpc_slice_unref(details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c index bda061ccb8..e6bcd9d6a3 100644 --- a/test/core/end2end/tests/cancel_after_accept.c +++ b/test/core/end2end/tests/cancel_after_accept.c @@ -81,9 +81,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -102,6 +103,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Cancel after accept, no payload */ @@ -250,8 +252,8 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, grpc_byte_buffer_destroy(response_payload_recv); grpc_slice_unref(details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); if (args != NULL) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; diff --git a/test/core/end2end/tests/cancel_after_client_done.c b/test/core/end2end/tests/cancel_after_client_done.c index 0e7291116f..d0e68354d9 100644 --- a/test/core/end2end/tests/cancel_after_client_done.c +++ b/test/core/end2end/tests/cancel_after_client_done.c @@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Cancel after accept with a writes closed, no payload */ @@ -227,8 +229,8 @@ static void test_cancel_after_accept_and_writes_closed( grpc_byte_buffer_destroy(response_payload_recv); grpc_slice_unref(details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); end_test(&f); diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c index 0ee59806fe..5bc9ed283b 100644 --- a/test/core/end2end/tests/cancel_after_invoke.c +++ b/test/core/end2end/tests/cancel_after_invoke.c @@ -49,11 +49,12 @@ static void *tag(intptr_t t) { return (void *)t; } static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, const char *test_name, cancellation_mode mode, + size_t test_ops, grpc_channel_args *client_args, grpc_channel_args *server_args) { grpc_end2end_test_fixture f; - gpr_log(GPR_INFO, "Running test: %s/%s/%s", test_name, config.name, - mode.name); + gpr_log(GPR_INFO, "Running test: %s/%s/%s [%" PRIdPTR " ops]", test_name, + config.name, mode.name, test_ops); f = config.create_fixture(client_args, server_args); config.init_server(&f, server_args); config.init_client(&f, client_args); @@ -78,9 +79,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) - .type == GRPC_OP_COMPLETE); + grpc_event ev = grpc_completion_queue_next( + f->cq, grpc_timeout_seconds_to_deadline(5), NULL); + GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); + GPR_ASSERT(ev.tag == tag(1000)); grpc_server_destroy(f->server); f->server = NULL; } @@ -98,6 +100,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Cancel after invoke, no payload */ @@ -106,8 +109,8 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_op ops[6]; grpc_op *op; grpc_call *c; - grpc_end2end_test_fixture f = - begin_test(config, "test_cancel_after_invoke", mode, NULL, NULL); + grpc_end2end_test_fixture f = begin_test(config, "test_cancel_after_invoke", + mode, test_ops, NULL, NULL); cq_verifier *cqv = cq_verifier_create(f.cq); grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; @@ -187,7 +190,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_byte_buffer_destroy(response_payload_recv); grpc_slice_unref(details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c index 7d53091a1a..67d2e9b8ab 100644 --- a/test/core/end2end/tests/cancel_before_invoke.c +++ b/test/core/end2end/tests/cancel_before_invoke.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Cancel before invoke */ @@ -184,7 +186,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, grpc_byte_buffer_destroy(response_payload_recv); grpc_slice_unref(details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); diff --git a/test/core/end2end/tests/cancel_in_a_vacuum.c b/test/core/end2end/tests/cancel_in_a_vacuum.c index c13fc88c53..1235cef2dc 100644 --- a/test/core/end2end/tests/cancel_in_a_vacuum.c +++ b/test/core/end2end/tests/cancel_in_a_vacuum.c @@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Cancel and do nothing */ @@ -116,7 +118,7 @@ static void test_cancel_in_a_vacuum(grpc_end2end_test_config config, GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c, NULL)); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(v_client); end_test(&f); diff --git a/test/core/end2end/tests/cancel_with_status.c b/test/core/end2end/tests/cancel_with_status.c index 1e0e33b4e2..e8259f99f1 100644 --- a/test/core/end2end/tests/cancel_with_status.c +++ b/test/core/end2end/tests/cancel_with_status.c @@ -77,9 +77,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) - .type == GRPC_OP_COMPLETE); + grpc_event ev = grpc_completion_queue_next( + f->cq, grpc_timeout_seconds_to_deadline(5), NULL); + GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); + GPR_ASSERT(ev.tag == tag(1000)); grpc_server_destroy(f->server); f->server = NULL; } @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -163,7 +165,7 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&initial_metadata_recv); grpc_metadata_array_destroy(&trailing_metadata_recv); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/compressed_payload.c b/test/core/end2end/tests/compressed_payload.c index c5b750af52..1fe8613adb 100644 --- a/test/core/end2end/tests/compressed_payload.c +++ b/test/core/end2end/tests/compressed_payload.c @@ -82,9 +82,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -103,6 +104,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void request_for_disabled_algorithm( @@ -259,8 +261,8 @@ static void request_for_disabled_algorithm( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); @@ -517,8 +519,8 @@ static void request_with_payload_template( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/connectivity.c b/test/core/end2end/tests/connectivity.c index 979419a100..eb84aaed16 100644 --- a/test/core/end2end/tests/connectivity.c +++ b/test/core/end2end/tests/connectivity.c @@ -171,6 +171,9 @@ static void test_connectivity(grpc_end2end_test_config config) { grpc_channel_destroy(f.client); grpc_completion_queue_shutdown(f.cq); grpc_completion_queue_destroy(f.cq); + + /* shutdown_cq is not used in this test */ + grpc_completion_queue_destroy(f.shutdown_cq); config.tear_down_data(&f); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/default_host.c b/test/core/end2end/tests/default_host.c index 8dccca20c7..efd2682932 100644 --- a/test/core/end2end/tests/default_host.c +++ b/test/core/end2end/tests/default_host.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_fixture f) { @@ -212,8 +214,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c index e2b888a3bb..23016f9ee5 100644 --- a/test/core/end2end/tests/disappearing_server.c +++ b/test/core/end2end/tests/disappearing_server.c @@ -79,6 +79,9 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + + /* Note: shutdown_cq was unused in this test */ + grpc_completion_queue_destroy(f->shutdown_cq); } static void do_request_and_shutdown_server(grpc_end2end_test_config config, @@ -188,8 +191,8 @@ static void do_request_and_shutdown_server(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); } static void disappearing_server_test(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/empty_batch.c b/test/core/end2end/tests/empty_batch.c index 37ca81384f..d0971367a6 100644 --- a/test/core/end2end/tests/empty_batch.c +++ b/test/core/end2end/tests/empty_batch.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void empty_batch_body(grpc_end2end_test_config config, @@ -119,7 +121,7 @@ static void empty_batch_body(grpc_end2end_test_config config, CQ_EXPECT_COMPLETION(cqv, tag(1), 1); cq_verify(cqv); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/filter_call_init_fails.c b/test/core/end2end/tests/filter_call_init_fails.c index 0c4f0dd42f..3402584fd9 100644 --- a/test/core/end2end/tests/filter_call_init_fails.c +++ b/test/core/end2end/tests/filter_call_init_fails.c @@ -84,9 +84,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -105,6 +106,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } // Simple request via a SERVER_CHANNEL filter that always fails to @@ -192,7 +194,7 @@ static void test_server_channel_filter(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); @@ -282,7 +284,7 @@ static void test_client_channel_filter(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); @@ -370,7 +372,7 @@ static void test_client_subchannel_filter(grpc_end2end_test_config config) { // Reset and create a new call. (The first call uses a different code // path in client_channel.c than subsequent calls on the same channel, // and we need to test both.) - grpc_call_destroy(c); + grpc_call_unref(c); status = GRPC_STATUS_OK; grpc_slice_unref(details); details = grpc_empty_slice(); @@ -397,7 +399,7 @@ static void test_client_subchannel_filter(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/filter_causes_close.c b/test/core/end2end/tests/filter_causes_close.c index 9115823a74..ede8fb49df 100644 --- a/test/core/end2end/tests/filter_causes_close.c +++ b/test/core/end2end/tests/filter_causes_close.c @@ -79,9 +79,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -100,6 +101,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Simple request via a server filter that always closes the stream.*/ @@ -187,7 +189,7 @@ static void test_request(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/filter_latency.c b/test/core/end2end/tests/filter_latency.c index d38ce22e09..7d2614a067 100644 --- a/test/core/end2end/tests/filter_latency.c +++ b/test/core/end2end/tests/filter_latency.c @@ -86,9 +86,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -107,6 +108,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } // Simple request via a server filter that saves the reported latency value. @@ -226,8 +228,8 @@ static void test_request(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(s); - grpc_call_destroy(c); + grpc_call_unref(s); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c index 182bb20780..37ac33818f 100644 --- a/test/core/end2end/tests/graceful_server_shutdown.c +++ b/test/core/end2end/tests/graceful_server_shutdown.c @@ -91,6 +91,8 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + /* Note: shutdown_cq is not used in this test */ + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_early_server_shutdown_finishes_inflight_calls( @@ -190,7 +192,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( CQ_EXPECT_COMPLETION(cqv, tag(1), 1); cq_verify(cqv); - grpc_call_destroy(s); + grpc_call_unref(s); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo")); @@ -204,7 +206,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/high_initial_seqno.c b/test/core/end2end/tests/high_initial_seqno.c index f534a19c87..893fdd6862 100644 --- a/test/core/end2end/tests/high_initial_seqno.c +++ b/test/core/end2end/tests/high_initial_seqno.c @@ -78,9 +78,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -99,6 +100,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -203,8 +205,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); /* TODO(ctiller): this rate limits the test, and it should be removed when retry has been implemented; until then cross-thread chatter diff --git a/test/core/end2end/tests/hpack_size.c b/test/core/end2end/tests/hpack_size.c index 17ef204ce9..b1db58271e 100644 --- a/test/core/end2end/tests/hpack_size.c +++ b/test/core/end2end/tests/hpack_size.c @@ -218,9 +218,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -239,6 +240,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -356,8 +358,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/idempotent_request.c b/test/core/end2end/tests/idempotent_request.c index f80654db1a..1ea2ac4774 100644 --- a/test/core/end2end/tests/idempotent_request.c +++ b/test/core/end2end/tests/idempotent_request.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -217,8 +219,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c index f4b93d9873..a49cf4f3d7 100644 --- a/test/core/end2end/tests/invoke_large_request.c +++ b/test/core/end2end/tests/invoke_large_request.c @@ -71,9 +71,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -92,6 +93,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static grpc_slice large_slice(void) { @@ -256,8 +258,8 @@ static void test_invoke_large_request(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/keepalive_timeout.c b/test/core/end2end/tests/keepalive_timeout.c index bf6ca0d9d9..ceefe5db8e 100644 --- a/test/core/end2end/tests/keepalive_timeout.c +++ b/test/core/end2end/tests/keepalive_timeout.c @@ -78,8 +78,9 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000), + + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), five_seconds_from_now(), NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); @@ -99,6 +100,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request, server replies with a payload, then waits for the @@ -188,8 +190,6 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == error); CQ_EXPECT_COMPLETION(cqv, tag(102), 1); - cq_verify(cqv); - CQ_EXPECT_COMPLETION(cqv, tag(1), 1); cq_verify(cqv); @@ -223,8 +223,8 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/large_metadata.c b/test/core/end2end/tests/large_metadata.c index a670e411e9..dd796889db 100644 --- a/test/core/end2end/tests/large_metadata.c +++ b/test/core/end2end/tests/large_metadata.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } // Request with a large amount of metadata. @@ -244,8 +246,8 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/load_reporting_hook.c b/test/core/end2end/tests/load_reporting_hook.c index ea3dd2368b..5f671405c2 100644 --- a/test/core/end2end/tests/load_reporting_hook.c +++ b/test/core/end2end/tests/load_reporting_hook.c @@ -101,9 +101,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -122,13 +123,13 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void request_response_with_payload( grpc_end2end_test_config config, grpc_end2end_test_fixture f, const char *method_name, const char *request_msg, const char *response_msg, - grpc_metadata *initial_lr_metadata, - grpc_load_reporting_cost_context *cost_ctx) { + grpc_metadata *initial_lr_metadata, grpc_metadata *trailing_lr_metadata) { grpc_slice request_payload_slice = grpc_slice_from_static_string(request_msg); grpc_slice response_payload_slice = grpc_slice_from_static_string(response_msg); @@ -241,8 +242,9 @@ static void request_response_with_payload( op->reserved = NULL; op++; op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; - GPR_ASSERT(cost_ctx != NULL); - grpc_call_set_load_reporting_cost_context(s, cost_ctx); + GPR_ASSERT(trailing_lr_metadata != NULL); + op->data.send_status_from_server.trailing_metadata_count = 1; + op->data.send_status_from_server.trailing_metadata = trailing_lr_metadata; op->data.send_status_from_server.status = GRPC_STATUS_OK; grpc_slice status_details = grpc_slice_from_static_string("xyz"); op->data.send_status_from_server.status_details = &status_details; @@ -264,8 +266,8 @@ static void request_response_with_payload( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); @@ -296,21 +298,21 @@ static void test_load_reporting_hook(grpc_end2end_test_config config) { const char *response_msg = "... and the response from the server"; grpc_metadata initial_lr_metadata; + grpc_metadata trailing_lr_metadata; initial_lr_metadata.key = GRPC_MDSTR_LB_TOKEN; initial_lr_metadata.value = grpc_slice_from_static_string("client-token"); memset(&initial_lr_metadata.internal_data, 0, sizeof(initial_lr_metadata.internal_data)); - grpc_load_reporting_cost_context *cost_ctx = gpr_malloc(sizeof(*cost_ctx)); - memset(cost_ctx, 0, sizeof(*cost_ctx)); - cost_ctx->values_count = 1; - cost_ctx->values = - gpr_malloc(sizeof(*cost_ctx->values) * cost_ctx->values_count); - cost_ctx->values[0] = grpc_slice_from_static_string("cost-token"); + trailing_lr_metadata.key = GRPC_MDSTR_LB_COST_BIN; + trailing_lr_metadata.value = grpc_slice_from_static_string("server-token"); + memset(&trailing_lr_metadata.internal_data, 0, + sizeof(trailing_lr_metadata.internal_data)); request_response_with_payload(config, f, method_name, request_msg, - response_msg, &initial_lr_metadata, cost_ctx); + response_msg, &initial_lr_metadata, + &trailing_lr_metadata); end_test(&f); { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c index 2360138ede..8ec92e476a 100644 --- a/test/core/end2end/tests/max_concurrent_streams.c +++ b/test/core/end2end/tests/max_concurrent_streams.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -199,8 +201,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } @@ -431,10 +433,10 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { cq_verifier_destroy(cqv); - grpc_call_destroy(c1); - grpc_call_destroy(s1); - grpc_call_destroy(c2); - grpc_call_destroy(s2); + grpc_call_unref(c1); + grpc_call_unref(s1); + grpc_call_unref(c2); + grpc_call_unref(s2); grpc_slice_unref(details1); grpc_slice_unref(details2); @@ -626,10 +628,10 @@ static void test_max_concurrent_streams_with_timeout_on_first( cq_verifier_destroy(cqv); - grpc_call_destroy(c1); - grpc_call_destroy(s1); - grpc_call_destroy(c2); - grpc_call_destroy(s2); + grpc_call_unref(c1); + grpc_call_unref(s1); + grpc_call_unref(c2); + grpc_call_unref(s2); grpc_slice_unref(details1); grpc_slice_unref(details2); @@ -787,7 +789,7 @@ static void test_max_concurrent_streams_with_timeout_on_second( /* second request is finished because of time out, so destroy the second call */ - grpc_call_destroy(c2); + grpc_call_unref(c2); /* now reply the first call */ memset(ops, 0, sizeof(ops)); @@ -819,8 +821,8 @@ static void test_max_concurrent_streams_with_timeout_on_second( cq_verifier_destroy(cqv); - grpc_call_destroy(c1); - grpc_call_destroy(s1); + grpc_call_unref(c1); + grpc_call_unref(s1); grpc_slice_unref(details1); grpc_slice_unref(details2); diff --git a/test/core/end2end/tests/max_connection_age.c b/test/core/end2end/tests/max_connection_age.c index 04bdb39445..c470dcda69 100644 --- a/test/core/end2end/tests/max_connection_age.c +++ b/test/core/end2end/tests/max_connection_age.c @@ -89,6 +89,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_max_age_forcibly_close(grpc_end2end_test_config config) { @@ -213,7 +214,7 @@ static void test_max_age_forcibly_close(grpc_end2end_test_config config) { CQ_EXPECT_COMPLETION(cqv, tag(0xdead), true); cq_verify(cqv); - grpc_call_destroy(s); + grpc_call_unref(s); /* The connection should be closed immediately after the max age grace period, the in-progress RPC should fail. */ @@ -229,7 +230,7 @@ static void test_max_age_forcibly_close(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&trailing_metadata_recv); grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); @@ -351,7 +352,7 @@ static void test_max_age_gracefully_close(grpc_end2end_test_config config) { CQ_EXPECT_COMPLETION(cqv, tag(0xdead), true); cq_verify(cqv); - grpc_call_destroy(s); + grpc_call_unref(s); /* The connection is closed gracefully with goaway, the rpc should still be completed. */ @@ -367,7 +368,7 @@ static void test_max_age_gracefully_close(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&trailing_metadata_recv); grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/max_connection_idle.c b/test/core/end2end/tests/max_connection_idle.c index 98bc08c6d5..488d945123 100644 --- a/test/core/end2end/tests/max_connection_idle.c +++ b/test/core/end2end/tests/max_connection_idle.c @@ -175,8 +175,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } @@ -242,6 +242,7 @@ static void test_max_connection_idle(grpc_end2end_test_config config) { grpc_completion_queue_shutdown(f.cq); drain_cq(f.cq); grpc_completion_queue_destroy(f.cq); + grpc_completion_queue_destroy(f.shutdown_cq); config.tear_down_data(&f); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c index a8b6f1f79a..f65edab865 100644 --- a/test/core/end2end/tests/max_message_length.c +++ b/test/core/end2end/tests/max_message_length.c @@ -83,9 +83,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -104,6 +105,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } // Test with request larger than the limit. @@ -287,8 +289,8 @@ done: grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(recv_payload); - grpc_call_destroy(c); - if (s != NULL) grpc_call_destroy(s); + grpc_call_unref(c); + if (s != NULL) grpc_call_unref(s); cq_verifier_destroy(cqv); @@ -481,8 +483,8 @@ static void test_max_message_length_on_response(grpc_end2end_test_config config, grpc_byte_buffer_destroy(response_payload); grpc_byte_buffer_destroy(recv_payload); - grpc_call_destroy(c); - if (s != NULL) grpc_call_destroy(s); + grpc_call_unref(c); + if (s != NULL) grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/negative_deadline.c b/test/core/end2end/tests/negative_deadline.c index e7448bd6bc..113b650a97 100644 --- a/test/core/end2end/tests/negative_deadline.c +++ b/test/core/end2end/tests/negative_deadline.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -160,7 +162,7 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&initial_metadata_recv); grpc_metadata_array_destroy(&trailing_metadata_recv); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/network_status_change.c b/test/core/end2end/tests/network_status_change.c index 1a5f2b25cc..1ea4be81b1 100644 --- a/test/core/end2end/tests/network_status_change.c +++ b/test/core/end2end/tests/network_status_change.c @@ -77,9 +77,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -98,6 +99,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -229,8 +231,8 @@ static void test_invoke_network_status_change(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/no_logging.c b/test/core/end2end/tests/no_logging.c index 683d6465db..17b6886f2c 100644 --- a/test/core/end2end/tests/no_logging.c +++ b/test/core/end2end/tests/no_logging.c @@ -104,9 +104,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -125,6 +126,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -242,8 +244,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/no_op.c b/test/core/end2end/tests/no_op.c index 698d8fbc60..0d98fc9c0a 100644 --- a/test/core/end2end/tests/no_op.c +++ b/test/core/end2end/tests/no_op.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_no_op(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/payload.c b/test/core/end2end/tests/payload.c index 5751b801b5..3f25ecb8ff 100644 --- a/test/core/end2end/tests/payload.c +++ b/test/core/end2end/tests/payload.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Creates and returns a grpc_slice containing random alphanumeric characters. @@ -259,8 +261,8 @@ static void request_response_with_payload(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/ping.c b/test/core/end2end/tests/ping.c index c51c4993e2..027c17ea4c 100644 --- a/test/core/end2end/tests/ping.c +++ b/test/core/end2end/tests/ping.c @@ -112,6 +112,9 @@ static void test_ping(grpc_end2end_test_config config, grpc_channel_destroy(f.client); grpc_completion_queue_shutdown(f.cq); grpc_completion_queue_destroy(f.cq); + + /* f.shutdown_cq is not used in this test */ + grpc_completion_queue_destroy(f.shutdown_cq); config.tear_down_data(&f); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c index cd81ce24fe..598e45ee33 100644 --- a/test/core/end2end/tests/ping_pong_streaming.c +++ b/test/core/end2end/tests/ping_pong_streaming.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client pings and server pongs. Repeat messages rounds before finishing. */ @@ -263,8 +265,8 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, CQ_EXPECT_COMPLETION(cqv, tag(104), 1); cq_verify(cqv); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c index 242add37cb..7e07a7129c 100644 --- a/test/core/end2end/tests/registered_call.c +++ b/test/core/end2end/tests/registered_call.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -198,8 +200,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/request_with_flags.c b/test/core/end2end/tests/request_with_flags.c index 10ecf5b1af..daf15fa5f7 100644 --- a/test/core/end2end/tests/request_with_flags.c +++ b/test/core/end2end/tests/request_with_flags.c @@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_invoke_request_with_flags( @@ -177,7 +179,7 @@ static void test_invoke_request_with_flags( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); + grpc_call_unref(c); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c index 4eb3399bb9..43686d58ab 100644 --- a/test/core/end2end/tests/request_with_payload.c +++ b/test/core/end2end/tests/request_with_payload.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -224,8 +226,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/resource_quota_server.c b/test/core/end2end/tests/resource_quota_server.c index 4d3ce1c937..6418f5d01a 100644 --- a/test/core/end2end/tests/resource_quota_server.c +++ b/test/core/end2end/tests/resource_quota_server.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Creates and returns a grpc_slice containing random alphanumeric characters. @@ -167,6 +169,7 @@ void resource_quota_server(grpc_end2end_test_config config) { int cancelled_calls_on_client = 0; int cancelled_calls_on_server = 0; int deadline_exceeded = 0; + int unavailable = 0; grpc_byte_buffer *request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); @@ -203,7 +206,7 @@ void resource_quota_server(grpc_end2end_test_config config) { op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; op->data.send_initial_metadata.count = 0; - op->flags = 0; + op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY; op->reserved = NULL; op++; op->op = GRPC_OP_SEND_MESSAGE; @@ -258,6 +261,9 @@ void resource_quota_server(grpc_end2end_test_config config) { case GRPC_STATUS_DEADLINE_EXCEEDED: deadline_exceeded++; break; + case GRPC_STATUS_UNAVAILABLE: + unavailable++; + break; case GRPC_STATUS_OK: break; default: @@ -268,7 +274,7 @@ void resource_quota_server(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&initial_metadata_recv[call_id]); grpc_metadata_array_destroy(&trailing_metadata_recv[call_id]); - grpc_call_destroy(client_calls[call_id]); + grpc_call_unref(client_calls[call_id]); grpc_slice_unref(details[call_id]); pending_client_calls--; @@ -350,15 +356,15 @@ void resource_quota_server(grpc_end2end_test_config config) { GPR_ASSERT(pending_server_end_calls > 0); pending_server_end_calls--; - grpc_call_destroy(server_calls[call_id]); + grpc_call_unref(server_calls[call_id]); } } gpr_log(GPR_INFO, "Done. %d total calls: %d cancelled at server, %d cancelled at " - "client, %d timed out.", + "client, %d timed out, %d unavailable.", NUM_CALLS, cancelled_calls_on_server, cancelled_calls_on_client, - deadline_exceeded); + deadline_exceeded, unavailable); grpc_byte_buffer_destroy(request_payload); grpc_slice_unref(request_payload_slice); diff --git a/test/core/end2end/tests/server_finishes_request.c b/test/core/end2end/tests/server_finishes_request.c index 88bc90e778..be0901bfdd 100644 --- a/test/core/end2end/tests/server_finishes_request.c +++ b/test/core/end2end/tests/server_finishes_request.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -197,8 +199,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/shutdown_finishes_calls.c b/test/core/end2end/tests/shutdown_finishes_calls.c index 6f9421d19a..d19af1a74d 100644 --- a/test/core/end2end/tests/shutdown_finishes_calls.c +++ b/test/core/end2end/tests/shutdown_finishes_calls.c @@ -84,6 +84,8 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + /* f->shutdown_cq is not used in this test */ + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_early_server_shutdown_finishes_inflight_calls( @@ -184,8 +186,8 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/shutdown_finishes_tags.c b/test/core/end2end/tests/shutdown_finishes_tags.c index 51a242337f..008f72e6f4 100644 --- a/test/core/end2end/tests/shutdown_finishes_tags.c +++ b/test/core/end2end/tests/shutdown_finishes_tags.c @@ -84,6 +84,8 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + /* f->shutdown_cq is not used in this test */ + grpc_completion_queue_destroy(f->shutdown_cq); } static void test_early_server_shutdown_finishes_tags( diff --git a/test/core/end2end/tests/simple_cacheable_request.c b/test/core/end2end/tests/simple_cacheable_request.c index 4f1013a030..cc0f89c422 100644 --- a/test/core/end2end/tests/simple_cacheable_request.c +++ b/test/core/end2end/tests/simple_cacheable_request.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Request/response with metadata and payload.*/ @@ -272,8 +274,8 @@ static void test_cacheable_request_response_with_metadata_and_payload( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c index 978dff0dc0..9e938f86a6 100644 --- a/test/core/end2end/tests/simple_delayed_request.c +++ b/test/core/end2end/tests/simple_delayed_request.c @@ -62,9 +62,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -83,6 +84,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_delayed_request_body(grpc_end2end_test_config config, @@ -193,8 +195,8 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/simple_metadata.c b/test/core/end2end/tests/simple_metadata.c index 83d1037e4b..08679a588a 100644 --- a/test/core/end2end/tests/simple_metadata.c +++ b/test/core/end2end/tests/simple_metadata.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Request/response with metadata and payload.*/ @@ -264,8 +266,8 @@ static void test_request_response_with_metadata_and_payload( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c index 31bfef4cc2..45c717363b 100644 --- a/test/core/end2end/tests/simple_request.c +++ b/test/core/end2end/tests/simple_request.c @@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } static void simple_request_body(grpc_end2end_test_config config, @@ -217,8 +219,8 @@ static void simple_request_body(grpc_end2end_test_config config, grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); } diff --git a/test/core/end2end/tests/streaming_error_response.c b/test/core/end2end/tests/streaming_error_response.c index c652d9469d..e59675a655 100644 --- a/test/core/end2end/tests/streaming_error_response.c +++ b/test/core/end2end/tests/streaming_error_response.c @@ -79,9 +79,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -100,6 +101,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -182,6 +184,9 @@ static void test(grpc_end2end_test_config config, bool request_status_early) { GPR_ASSERT(GRPC_CALL_OK == error); CQ_EXPECT_COMPLETION(cqv, tag(102), 1); + if (!request_status_early) { + CQ_EXPECT_COMPLETION(cqv, tag(1), 1); + } cq_verify(cqv); memset(ops, 0, sizeof(ops)); @@ -193,9 +198,6 @@ static void test(grpc_end2end_test_config config, bool request_status_early) { GPR_ASSERT(GRPC_CALL_OK == error); CQ_EXPECT_COMPLETION(cqv, tag(103), 1); - if (!request_status_early) { - CQ_EXPECT_COMPLETION(cqv, tag(1), 1); - } cq_verify(cqv); memset(ops, 0, sizeof(ops)); @@ -261,8 +263,8 @@ static void test(grpc_end2end_test_config config, bool request_status_early) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/trailing_metadata.c b/test/core/end2end/tests/trailing_metadata.c index 3cbffdfe75..ca8eb6389e 100644 --- a/test/core/end2end/tests/trailing_metadata.c +++ b/test/core/end2end/tests/trailing_metadata.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Request/response with metadata and payload.*/ @@ -274,8 +276,8 @@ static void test_request_response_with_metadata_and_payload( grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/write_buffering.c b/test/core/end2end/tests/write_buffering.c index 2d4142b97d..f8f2102e67 100644 --- a/test/core/end2end/tests/write_buffering.c +++ b/test/core/end2end/tests/write_buffering.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -272,8 +274,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); diff --git a/test/core/end2end/tests/write_buffering_at_end.c b/test/core/end2end/tests/write_buffering_at_end.c index e61343e684..2facd9c1a4 100644 --- a/test/core/end2end/tests/write_buffering_at_end.c +++ b/test/core/end2end/tests/write_buffering_at_end.c @@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck( - f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + NULL) .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; @@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_shutdown(f->cq); drain_cq(f->cq); grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -263,8 +265,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); - grpc_call_destroy(c); - grpc_call_destroy(s); + grpc_call_unref(c); + grpc_call_unref(s); cq_verifier_destroy(cqv); |