diff options
Diffstat (limited to 'test')
60 files changed, 1319 insertions, 1270 deletions
diff --git a/test/core/end2end/cq_verifier.c b/test/core/end2end/cq_verifier.c index 8a30e012e9..33f7c02b61 100644 --- a/test/core/end2end/cq_verifier.c +++ b/test/core/end2end/cq_verifier.c @@ -167,6 +167,9 @@ static void verify_matches(expectation *e, grpc_event *ev) { static void expectation_to_strvec(gpr_strvec *buf, expectation *e) { char *tmp; + gpr_asprintf(&tmp, "%p ", e->tag); + gpr_strvec_add(buf, tmp); + switch (e->type) { case GRPC_OP_COMPLETE: gpr_asprintf(&tmp, "GRPC_OP_COMPLETE result=%d", e->success); diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index 06cbba295e..7d3568c22e 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -62,12 +62,10 @@ void test_connect(const char *server_host, const char *client_host, int port, char *server_hostport; grpc_channel *client; grpc_server *server; - grpc_completion_queue *client_cq; - grpc_completion_queue *server_cq; + grpc_completion_queue *cq; grpc_call *c; grpc_call *s; - cq_verifier *v_client; - cq_verifier *v_server; + cq_verifier *cqv; gpr_timespec deadline; int got_port; grpc_op ops[6]; @@ -93,9 +91,9 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_call_details_init(&call_details); /* Create server. */ - server_cq = grpc_completion_queue_create(); + cq = grpc_completion_queue_create(); server = grpc_server_create(NULL); - grpc_server_register_completion_queue(server, server_cq); + grpc_server_register_completion_queue(server, cq); GPR_ASSERT((got_port = grpc_server_add_http2_port(server, server_hostport)) > 0); if (port == 0) { @@ -104,13 +102,11 @@ void test_connect(const char *server_host, const char *client_host, int port, GPR_ASSERT(port == got_port); } grpc_server_start(server); - v_server = cq_verifier_create(server_cq); + cqv = cq_verifier_create(cq); /* Create client. */ gpr_join_host_port(&client_hostport, client_host, port); - client_cq = grpc_completion_queue_create(); client = grpc_channel_create(client_hostport, NULL); - v_client = cq_verifier_create(client_cq); gpr_log(GPR_INFO, "Testing with server=%s client=%s (expecting %s)", server_hostport, client_hostport, expect_ok ? "success" : "failure"); @@ -128,7 +124,7 @@ void test_connect(const char *server_host, const char *client_host, int port, } /* Send a trivial request. */ - c = grpc_channel_create_call(client, client_cq, "/foo", "foo.test.google.fr", + c = grpc_channel_create_call(client, cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -155,12 +151,11 @@ void test_connect(const char *server_host, const char *client_host, int port, if (expect_ok) { /* Check for a successful request. */ - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(server, &s, &call_details, - &request_metadata_recv, server_cq, - server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + server, &s, &call_details, + &request_metadata_recv, cq, cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -180,45 +175,41 @@ void test_connect(const char *server_host, const char *client_host, int port, GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); grpc_call_destroy(s); } else { /* Check for a failed connection. */ - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED); } grpc_call_destroy(c); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); /* Destroy client. */ grpc_channel_destroy(client); - grpc_completion_queue_shutdown(client_cq); - drain_cq(client_cq); - grpc_completion_queue_destroy(client_cq); /* Destroy server. */ - grpc_server_shutdown_and_notify(server, server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + grpc_server_shutdown_and_notify(server, cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(cq, tag(1000), + GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(server); - grpc_completion_queue_shutdown(server_cq); - drain_cq(server_cq); - grpc_completion_queue_destroy(server_cq); + grpc_completion_queue_shutdown(cq); + drain_cq(cq); + grpc_completion_queue_destroy(cq); grpc_call_details_destroy(&call_details); gpr_free(details); diff --git a/test/core/end2end/end2end_tests.h b/test/core/end2end/end2end_tests.h index a61c725aa2..a18c702951 100644 --- a/test/core/end2end/end2end_tests.h +++ b/test/core/end2end/end2end_tests.h @@ -44,8 +44,7 @@ typedef struct grpc_end2end_test_config grpc_end2end_test_config; #define FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS 4 struct grpc_end2end_test_fixture { - grpc_completion_queue *server_cq; - grpc_completion_queue *client_cq; + grpc_completion_queue *cq; grpc_server *server; grpc_channel *client; void *fixture_data; @@ -65,4 +64,4 @@ struct grpc_end2end_test_config { void grpc_end2end_tests(grpc_end2end_test_config config); -#endif /* GRPC_TEST_CORE_END2END_END2END_TESTS_H */ +#endif /* GRPC_TEST_CORE_END2END_END2END_TESTS_H */ diff --git a/test/core/end2end/fixtures/chttp2_fake_security.c b/test/core/end2end/fixtures/chttp2_fake_security.c index 5323e29e82..f879b43f79 100644 --- a/test/core/end2end/fixtures/chttp2_fake_security.c +++ b/test/core/end2end/fixtures/chttp2_fake_security.c @@ -60,8 +60,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -83,8 +82,9 @@ static void chttp2_init_server_secure_fullstack( grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); - GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds)); + grpc_server_register_completion_queue(f->server, f->cq); + GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, + server_creds)); grpc_server_credentials_release(server_creds); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack.c b/test/core/end2end/fixtures/chttp2_fullstack.c index f92b40efeb..b83e227a89 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack.c +++ b/test/core/end2end/fixtures/chttp2_fullstack.c @@ -65,8 +65,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -84,7 +83,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c index 02aa575065..94b54253e4 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c @@ -71,8 +71,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( unique++); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -90,7 +89,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c index f92b40efeb..00322d4011 100644 --- a/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c +++ b/test/core/end2end/fixtures/chttp2_fullstack_with_poll.c @@ -65,8 +65,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -84,7 +83,7 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f, grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); grpc_server_start(f->server); } @@ -105,6 +104,8 @@ static grpc_end2end_test_config configs[] = { int main(int argc, char **argv) { size_t i; + grpc_platform_become_multipoller = grpc_poll_become_multipoller; + grpc_test_init(argc, argv); grpc_init(); diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c index 6d1b7b5ff0..237d072702 100644 --- a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c +++ b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c @@ -63,8 +63,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -86,8 +85,9 @@ static void chttp2_init_server_secure_fullstack( grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); - GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds)); + grpc_server_register_completion_queue(f->server, f->cq); + GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, + server_creds)); grpc_server_credentials_release(server_creds); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack_with_poll.c b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack_with_poll.c index a5865d37c8..ff5642642d 100644 --- a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack_with_poll.c +++ b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack_with_poll.c @@ -63,8 +63,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -86,7 +85,7 @@ static void chttp2_init_server_secure_fullstack( grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds)); grpc_server_credentials_release(server_creds); diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c index 4a15d502a5..d4bb5d3ef5 100644 --- a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c +++ b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c @@ -61,8 +61,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack( gpr_join_host_port(&ffd->localaddr, "localhost", port); f.fixture_data = ffd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); return f; } @@ -84,8 +83,9 @@ static void chttp2_init_server_secure_fullstack( grpc_server_destroy(f->server); } f->server = grpc_server_create(server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); - GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds)); + grpc_server_register_completion_queue(f->server, f->cq); + GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, + server_creds)); grpc_server_credentials_release(server_creds); grpc_server_start(f->server); } diff --git a/test/core/end2end/fixtures/chttp2_socket_pair.c b/test/core/end2end/fixtures/chttp2_socket_pair.c index 48c121c7c4..d84405224b 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair.c @@ -95,8 +95,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536); @@ -119,7 +118,7 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f, grpc_endpoint_pair *sfd = f->fixture_data; GPR_ASSERT(!f->server); f->server = grpc_server_create_from_filters(NULL, 0, server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); grpc_server_start(f->server); grpc_create_chttp2_transport(server_setup_transport, f, server_args, sfd->server, NULL, 0, grpc_mdctx_create(), 0); diff --git a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c index 1d2e6f51c1..ac8b5eb86d 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c @@ -95,8 +95,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); *sfd = grpc_iomgr_create_endpoint_pair("fixture", 1); @@ -119,7 +118,7 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f, grpc_endpoint_pair *sfd = f->fixture_data; GPR_ASSERT(!f->server); f->server = grpc_server_create_from_filters(NULL, 0, server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); grpc_server_start(f->server); grpc_create_chttp2_transport(server_setup_transport, f, server_args, sfd->server, NULL, 0, grpc_mdctx_create(), 0); diff --git a/test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c b/test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c index 0834987fbe..e160812fa3 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c @@ -96,8 +96,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( grpc_end2end_test_fixture f; memset(&f, 0, sizeof(f)); f.fixture_data = sfd; - f.client_cq = grpc_completion_queue_create(); - f.server_cq = grpc_completion_queue_create(); + f.cq = grpc_completion_queue_create(); *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536); @@ -120,7 +119,7 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f, grpc_endpoint_pair *sfd = f->fixture_data; GPR_ASSERT(!f->server); f->server = grpc_server_create_from_filters(NULL, 0, server_args); - grpc_server_register_completion_queue(f->server, f->server_cq); + grpc_server_register_completion_queue(f->server, f->cq); grpc_server_start(f->server); grpc_create_chttp2_transport(server_setup_transport, f, server_args, sfd->server, NULL, 0, grpc_mdctx_create(), 0); diff --git a/test/core/end2end/gen_build_json.py b/test/core/end2end/gen_build_json.py index 66447527e1..f47c92bc47 100755 --- a/test/core/end2end/gen_build_json.py +++ b/test/core/end2end/gen_build_json.py @@ -60,7 +60,7 @@ default_test_options = TestOptions(False, False) # maps test names to options END2END_TESTS = { 'bad_hostname': default_test_options, - 'cancel_after_accept': TestOptions(flaky=True, secure=False), + 'cancel_after_accept': default_test_options, 'cancel_after_accept_and_writes_closed': default_test_options, 'cancel_after_invoke': default_test_options, 'cancel_before_invoke': default_test_options, @@ -71,7 +71,7 @@ END2END_TESTS = { 'early_server_shutdown_finishes_tags': default_test_options, 'empty_batch': default_test_options, 'graceful_server_shutdown': default_test_options, - 'invoke_large_request': TestOptions(flaky=True, secure=False), + 'invoke_large_request': default_test_options, 'max_concurrent_streams': default_test_options, 'max_message_length': default_test_options, 'no_op': default_test_options, diff --git a/test/core/end2end/tests/bad_hostname.c b/test/core/end2end/tests/bad_hostname.c index 809c5bb7d4..2509ea06a0 100644 --- a/test/core/end2end/tests/bad_hostname.c +++ b/test/core/end2end/tests/bad_hostname.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,18 +94,15 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call *c; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -114,8 +113,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) { char *details = NULL; size_t details_capacity = 0; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "slartibartfast.local", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "slartibartfast.local", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -144,8 +143,8 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNAUTHENTICATED); @@ -157,7 +156,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); - cq_verifier_destroy(v_client); + cq_verifier_destroy(cqv); } static void test_invoke_simple_request(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c index cc5586bd1b..1cc6b2d147 100644 --- a/test/core/end2end/tests/cancel_after_accept.c +++ b/test/core/end2end/tests/cancel_after_accept.c @@ -75,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -91,12 +93,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Cancel after accept, no payload */ @@ -106,10 +105,10 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, grpc_op *op; grpc_call *c; grpc_call *s; - grpc_end2end_test_fixture f = begin_test(config, "cancel_after_accept", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "cancel_after_accept", NULL, NULL); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; grpc_metadata_array request_metadata_recv; @@ -127,8 +126,8 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, grpc_raw_byte_buffer_create(&response_payload_slice, 1); int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -162,12 +161,11 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(2))); - cq_expect_completion(v_server, tag(2), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(2))); + cq_expect_completion(cqv, tag(2), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_MESSAGE; @@ -190,11 +188,9 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c)); - cq_expect_completion(v_server, tag(3), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(3), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == mode.expect_status); GPR_ASSERT(0 == strcmp(details, mode.expect_details)); @@ -214,8 +210,7 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); } diff --git a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c index 9a2fcf68a4..015d437543 100644 --- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c +++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c @@ -75,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -91,12 +93,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Cancel after accept with a writes closed, no payload */ @@ -106,10 +105,10 @@ static void test_cancel_after_accept_and_writes_closed( grpc_op *op; grpc_call *c; grpc_call *s; - grpc_end2end_test_fixture f = begin_test(config, "test_cancel_after_accept_and_writes_closed", NULL, NULL); + grpc_end2end_test_fixture f = begin_test( + config, "test_cancel_after_accept_and_writes_closed", NULL, NULL); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; grpc_metadata_array request_metadata_recv; @@ -127,8 +126,8 @@ static void test_cancel_after_accept_and_writes_closed( grpc_raw_byte_buffer_create(&response_payload_slice, 1); int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -165,12 +164,11 @@ static void test_cancel_after_accept_and_writes_closed( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(2))); - cq_expect_completion(v_server, tag(2), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(2))); + cq_expect_completion(cqv, tag(2), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_MESSAGE; @@ -193,11 +191,9 @@ static void test_cancel_after_accept_and_writes_closed( GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c)); - cq_expect_completion(v_server, tag(3), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(3), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == mode.expect_status); GPR_ASSERT(0 == strcmp(details, mode.expect_details)); @@ -217,8 +213,7 @@ static void test_cancel_after_accept_and_writes_closed( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); } diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c index 9b3ad280cb..414ec706ce 100644 --- a/test/core/end2end/tests/cancel_after_invoke.c +++ b/test/core/end2end/tests/cancel_after_invoke.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,12 +94,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Cancel after invoke, no payload */ @@ -109,7 +108,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_end2end_test_fixture f = begin_test(config, "test_cancel_after_invoke", mode, NULL, NULL); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; grpc_metadata_array request_metadata_recv; @@ -122,8 +121,8 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_byte_buffer *request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -162,8 +161,8 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c)); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == mode.expect_status); GPR_ASSERT(0 == strcmp(details, mode.expect_details)); @@ -179,7 +178,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_call_destroy(c); - cq_verifier_destroy(v_client); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); } diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c index ca19c469f6..3cfe56eca2 100644 --- a/test/core/end2end/tests/cancel_before_invoke.c +++ b/test/core/end2end/tests/cancel_before_invoke.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Cancel before invoke */ @@ -104,9 +103,10 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, grpc_op ops[6]; grpc_op *op; grpc_call *c; - grpc_end2end_test_fixture f = begin_test(config, "cancel_before_invoke", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "cancel_before_invoke", NULL, NULL); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; grpc_metadata_array request_metadata_recv; @@ -119,8 +119,8 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, grpc_byte_buffer *request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == grpc_call_cancel(c)); @@ -159,8 +159,8 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, test_ops, tag(1))); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_CANCELLED); @@ -175,7 +175,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, grpc_call_destroy(c); - cq_verifier_destroy(v_client); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); } diff --git a/test/core/end2end/tests/cancel_in_a_vacuum.c b/test/core/end2end/tests/cancel_in_a_vacuum.c index 5d7d762e5f..8bffc3f4d3 100644 --- a/test/core/end2end/tests/cancel_in_a_vacuum.c +++ b/test/core/end2end/tests/cancel_in_a_vacuum.c @@ -75,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -91,24 +93,22 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Cancel and do nothing */ static void test_cancel_in_a_vacuum(grpc_end2end_test_config config, cancellation_mode mode) { grpc_call *c; - grpc_end2end_test_fixture f = begin_test(config, "test_cancel_in_a_vacuum", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_cancel_in_a_vacuum", NULL, NULL); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); + cq_verifier *v_client = cq_verifier_create(f.cq); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c)); diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c index ac2574c804..b414755cd1 100644 --- a/test/core/end2end/tests/census_simple_request.c +++ b/test/core/end2end/tests/census_simple_request.c @@ -65,8 +65,10 @@ static void *tag(gpr_intptr t) { return (void *)t; } static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown_and_notify(f->server, f->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -88,20 +90,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_body(grpc_end2end_test_fixture f) { grpc_call *c; grpc_call *s; gpr_timespec deadline = n_seconds_time(5); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -113,7 +111,7 @@ static void test_body(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -143,12 +141,11 @@ static void test_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -167,17 +164,15 @@ static void test_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -188,8 +183,7 @@ static void test_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_invoke_request_with_census( diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c index cf9b79f117..9acd18902a 100644 --- a/test/core/end2end/tests/disappearing_server.c +++ b/test/core/end2end/tests/disappearing_server.c @@ -76,17 +76,13 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, - cq_verifier *v_client, - cq_verifier *v_server) { + cq_verifier *cqv) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); @@ -101,7 +97,7 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f->client, f->client_cq, "/foo", + c = grpc_channel_create_call(f->client, f->cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -131,16 +127,16 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f->server, &s, &call_details, - &request_metadata_recv, f->server_cq, - f->server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f->server, &s, + &call_details, + &request_metadata_recv, + f->cq, f->cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); /* should be able to shut down the server early - and still complete the request */ - grpc_server_shutdown_and_notify(f->server, f->server_cq, tag(1000)); + grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -159,18 +155,16 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_expect_completion(v_server, tag(1000), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_expect_completion(cqv, tag(1000), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -184,23 +178,21 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, static void disappearing_server_test(grpc_end2end_test_config config) { grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); gpr_log(GPR_INFO, "%s/%s", "disappearing_server_test", config.name); config.init_client(&f, NULL); config.init_server(&f, NULL); - do_request_and_shutdown_server(&f, v_client, v_server); + do_request_and_shutdown_server(&f, cqv); /* now destroy and recreate the server */ config.init_server(&f, NULL); - do_request_and_shutdown_server(&f, v_client, v_server); + do_request_and_shutdown_server(&f, cqv); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c index 5826ababaa..adc59b4e94 100644 --- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c +++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c @@ -81,12 +81,9 @@ static void shutdown_client(grpc_end2end_test_fixture *f) { static void end_test(grpc_end2end_test_fixture *f) { shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_early_server_shutdown_finishes_inflight_calls( @@ -94,9 +91,9 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - grpc_end2end_test_fixture f = begin_test(config, "test_early_server_shutdown_finishes_inflight_calls", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = begin_test( + config, "test_early_server_shutdown_finishes_inflight_calls", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -108,8 +105,8 @@ static void test_early_server_shutdown_finishes_inflight_calls( size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -139,12 +136,11 @@ static void test_early_server_shutdown_finishes_inflight_calls( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -154,18 +150,16 @@ static void test_early_server_shutdown_finishes_inflight_calls( GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); /* shutdown and destroy the server */ - grpc_server_shutdown_and_notify(f.server, f.server_cq, tag(1000)); + grpc_server_shutdown_and_notify(f.server, f.cq, tag(1000)); grpc_server_cancel_all_calls(f.server); - cq_expect_completion(v_server, tag(102), 1); - cq_expect_completion(v_server, tag(1000), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(1000), 1); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); grpc_server_destroy(f.server); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); - GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); @@ -180,8 +174,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c index 0c2cdb78f0..fc03cb01a8 100644 --- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c +++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c @@ -81,18 +81,16 @@ static void shutdown_client(grpc_end2end_test_fixture *f) { static void end_test(grpc_end2end_test_fixture *f) { shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_early_server_shutdown_finishes_tags( grpc_end2end_test_config config) { - grpc_end2end_test_fixture f = begin_test(config, "test_early_server_shutdown_finishes_tags", NULL, NULL); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = begin_test( + config, "test_early_server_shutdown_finishes_tags", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_call *s = (void *)1; grpc_call_details call_details; grpc_metadata_array request_metadata_recv; @@ -102,21 +100,20 @@ static void test_early_server_shutdown_finishes_tags( /* upon shutdown, the server should finish all requested calls indicating no new call */ - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - grpc_server_shutdown_and_notify(f.server, f.server_cq, tag(1000)); - cq_expect_completion(v_server, tag(101), 0); - cq_expect_completion(v_server, tag(1000), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + grpc_server_shutdown_and_notify(f.server, f.cq, tag(1000)); + cq_expect_completion(cqv, tag(101), 0); + cq_expect_completion(cqv, tag(1000), 1); + cq_verify(cqv); GPR_ASSERT(s == NULL); grpc_server_destroy(f.server); end_test(&f); config.tear_down_data(&f); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } void grpc_end2end_tests(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/empty_batch.c b/test/core/end2end/tests/empty_batch.c index 79d1965981..db8458d3d9 100644 --- a/test/core/end2end/tests/empty_batch.c +++ b/test/core/end2end/tests/empty_batch.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,31 +94,28 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void empty_batch_body(grpc_end2end_test_fixture f) { grpc_call *c; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op *op = NULL; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, op, 0, tag(1))); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); grpc_call_destroy(c); - cq_verifier_destroy(v_client); + cq_verifier_destroy(cqv); } static void test_invoke_empty_body(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c index 6a5b4b15d3..8c1889add9 100644 --- a/test/core/end2end/tests/graceful_server_shutdown.c +++ b/test/core/end2end/tests/graceful_server_shutdown.c @@ -88,12 +88,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_early_server_shutdown_finishes_inflight_calls( @@ -101,9 +98,9 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - grpc_end2end_test_fixture f = begin_test(config, "test_early_server_shutdown_finishes_inflight_calls", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = begin_test( + config, "test_early_server_shutdown_finishes_inflight_calls", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -115,8 +112,8 @@ static void test_early_server_shutdown_finishes_inflight_calls( size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -146,16 +143,15 @@ static void test_early_server_shutdown_finishes_inflight_calls( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); /* shutdown and destroy the server */ - grpc_server_shutdown_and_notify(f.server, f.server_cq, tag(0xdead)); - cq_verify_empty(v_server); + grpc_server_shutdown_and_notify(f.server, f.cq, tag(0xdead)); + cq_verify_empty(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -174,19 +170,17 @@ static void test_early_server_shutdown_finishes_inflight_calls( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_expect_completion(v_server, tag(0xdead), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(0xdead), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); grpc_call_destroy(s); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); - GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -196,8 +190,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( grpc_call_destroy(c); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c index 812bdfa3e7..ae85af980a 100644 --- a/test/core/end2end/tests/invoke_large_request.c +++ b/test/core/end2end/tests/invoke_large_request.c @@ -72,8 +72,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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -88,12 +90,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static gpr_slice large_slice(void) { @@ -103,7 +102,8 @@ static gpr_slice large_slice(void) { } static void test_invoke_large_request(grpc_end2end_test_config config) { - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_large_request", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_invoke_large_request", NULL, NULL); gpr_slice request_payload_slice = large_slice(); gpr_slice response_payload_slice = large_slice(); @@ -114,8 +114,7 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { grpc_byte_buffer *response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1); gpr_timespec deadline = n_seconds_time(30); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -129,8 +128,8 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -141,71 +140,78 @@ static void test_invoke_large_request(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++; op->op = GRPC_OP_SEND_MESSAGE; op->data.send_message = request_payload; + op->flags = 0; op++; op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; op++; op->op = GRPC_OP_RECV_INITIAL_METADATA; op->data.recv_initial_metadata = &initial_metadata_recv; + op->flags = 0; op++; op->op = GRPC_OP_RECV_MESSAGE; op->data.recv_message = &response_payload_recv; + op->flags = 0; op++; op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; op->data.recv_status_on_client.status = &status; op->data.recv_status_on_client.status_details = &details; op->data.recv_status_on_client.status_details_capacity = &details_capacity; + op->flags = 0; op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; op->data.send_initial_metadata.count = 0; + op->flags = 0; op++; op->op = GRPC_OP_RECV_MESSAGE; op->data.recv_message = &request_payload_recv; + op->flags = 0; op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; op->data.recv_close_on_server.cancelled = &was_cancelled; + op->flags = 0; op++; op->op = GRPC_OP_SEND_MESSAGE; op->data.send_message = response_payload; + op->flags = 0; op++; - op = ops; op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; op->data.send_status_from_server.trailing_metadata_count = 0; op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED; op->data.send_status_from_server.status_details = "xyz"; + op->flags = 0; op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -216,8 +222,7 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c index f8660bb901..1204c070af 100644 --- a/test/core/end2end/tests/max_concurrent_streams.c +++ b/test/core/end2end/tests/max_concurrent_streams.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,20 +92,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -115,7 +113,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -145,12 +143,11 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -169,17 +166,15 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -190,8 +185,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_max_concurrent_streams(grpc_end2end_test_config config) { @@ -204,8 +198,7 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { grpc_call *s2; int live_call; gpr_timespec deadline; - cq_verifier *v_client; - cq_verifier *v_server; + cq_verifier *cqv; grpc_event ev; grpc_call_details call_details; grpc_metadata_array request_metadata_recv; @@ -222,6 +215,8 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { grpc_op ops[6]; grpc_op *op; int was_cancelled; + int got_client_start; + int got_server_start; server_arg.key = GRPC_ARG_MAX_CONCURRENT_STREAMS; server_arg.type = GRPC_ARG_INTEGER; @@ -231,8 +226,7 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { server_args.args = &server_arg; f = begin_test(config, "test_max_concurrent_streams", NULL, &server_args); - v_client = cq_verifier_create(f.client_cq); - v_server = cq_verifier_create(f.server_cq); + cqv = cq_verifier_create(f.cq); grpc_metadata_array_init(&request_metadata_recv); grpc_metadata_array_init(&initial_metadata_recv1); @@ -249,18 +243,17 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { /* start two requests - ensuring that the second is not accepted until the first completes */ - deadline = n_seconds_time(10); - c1 = grpc_channel_create_call(f.client, f.client_cq, "/alpha", + deadline = n_seconds_time(1000); + c1 = grpc_channel_create_call(f.client, f.cq, "/alpha", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c1); - c2 = grpc_channel_create_call(f.client, f.client_cq, "/beta", + c2 = grpc_channel_create_call(f.client, f.cq, "/beta", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c2); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s1, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s1, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -314,18 +307,28 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c2, ops, op - ops, tag(402))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); - - ev = grpc_completion_queue_next(f.client_cq, - GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3)); - GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); - GPR_ASSERT(ev.success); - GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401)); - /* The /alpha or /beta calls started above could be invoked (but NOT both); - * check this here */ - /* We'll get tag 303 or 403, we want 300, 400 */ - live_call = ((int)(gpr_intptr)ev.tag) - 1; + got_client_start = 0; + got_server_start = 0; + live_call = -1; + while (!got_client_start || !got_server_start) { + ev = grpc_completion_queue_next(f.cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3)); + GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); + GPR_ASSERT(ev.success); + if (ev.tag == tag(101)) { + GPR_ASSERT(!got_server_start); + got_server_start = 1; + } else { + GPR_ASSERT(!got_client_start); + GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401)); + /* The /alpha or /beta calls started above could be invoked (but NOT + * both); + * check this here */ + /* We'll get tag 303 or 403, we want 300, 400 */ + live_call = ((int)(gpr_intptr)ev.tag) - 1; + got_client_start = 1; + } + } + GPR_ASSERT(live_call == 300 || live_call == 400); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -345,21 +348,18 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s1, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(live_call + 2), 1); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(live_call + 2), 1); /* first request is finished, we should be able to start the second */ live_call = (live_call == 300) ? 400 : 300; - cq_expect_completion(v_client, tag(live_call + 1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(live_call + 1), 1); + cq_verify(cqv); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s2, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(201))); - cq_expect_completion(v_server, tag(201), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s2, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(201))); + cq_expect_completion(cqv, tag(201), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -379,14 +379,11 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s2, ops, op - ops, tag(202))); - cq_expect_completion(v_client, tag(live_call + 2), 1); - cq_verify(v_client); - - cq_expect_completion(v_server, tag(202), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(live_call + 2), 1); + cq_expect_completion(cqv, tag(202), 1); + cq_verify(cqv); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_call_destroy(c1); grpc_call_destroy(s1); diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c index 8d3d2f10d9..9d716f1135 100644 --- a/test/core/end2end/tests/max_message_length.c +++ b/test/core/end2end/tests/max_message_length.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_max_message_length(grpc_end2end_test_config config) { @@ -104,8 +103,7 @@ static void test_max_message_length(grpc_end2end_test_config config) { grpc_channel_args server_args; grpc_call *c; grpc_call *s; - cq_verifier *v_client; - cq_verifier *v_server; + cq_verifier *cqv; grpc_op ops[6]; grpc_op *op; gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world"); @@ -128,10 +126,9 @@ static void test_max_message_length(grpc_end2end_test_config config) { server_args.args = &server_arg; f = begin_test(config, "test_max_message_length", NULL, &server_args); - v_client = cq_verifier_create(f.client_cq); - v_server = cq_verifier_create(f.server_cq); + cqv = cq_verifier_create(f.cq); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", gpr_inf_future); GPR_ASSERT(c); @@ -165,12 +162,11 @@ static void test_max_message_length(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -179,11 +175,9 @@ static void test_max_message_length(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status != GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); @@ -200,8 +194,7 @@ static void test_max_message_length(grpc_end2end_test_config config) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); end_test(&f); config.tear_down_data(&f); diff --git a/test/core/end2end/tests/no_op.c b/test/core/end2end/tests/no_op.c index 1f32691e63..8ead664758 100644 --- a/test/core/end2end/tests/no_op.c +++ b/test/core/end2end/tests/no_op.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_no_op(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c index 2b617ce00b..8a3ec96212 100644 --- a/test/core/end2end/tests/ping_pong_streaming.c +++ b/test/core/end2end/tests/ping_pong_streaming.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,23 +92,20 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Client pings and server pongs. Repeat messages rounds before finishing. */ static void test_pingpong_streaming(grpc_end2end_test_config config, int messages) { - grpc_end2end_test_fixture f = begin_test(config, "test_pingpong_streaming", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_pingpong_streaming", NULL, NULL); grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -125,7 +124,7 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world"); gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you"); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -152,12 +151,11 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(100))); - cq_expect_completion(v_server, tag(100), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(100))); + cq_expect_completion(cqv, tag(100), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -192,8 +190,8 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_MESSAGE; @@ -202,11 +200,9 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(2), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(2), 1); + cq_verify(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); @@ -232,19 +228,16 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(104))); - cq_expect_completion(v_client, tag(1), 1); - cq_expect_completion(v_client, tag(3), 1); - cq_verify(v_client); - - cq_expect_completion(v_server, tag(101), 1); - cq_expect_completion(v_server, tag(104), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(1), 1); + cq_expect_completion(cqv, tag(3), 1); + cq_expect_completion(cqv, tag(101), 1); + cq_expect_completion(cqv, tag(104), 1); + cq_verify(cqv); grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_metadata_array_destroy(&initial_metadata_recv); grpc_metadata_array_destroy(&trailing_metadata_recv); diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c index 43e34ff673..f44fd3a224 100644 --- a/test/core/end2end/tests/registered_call.c +++ b/test/core/end2end/tests/registered_call.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,20 +94,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -117,7 +115,7 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_registered_call(f.client, f.client_cq, rc, deadline); + c = grpc_channel_create_registered_call(f.client, f.cq, rc, deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -146,12 +144,11 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -170,17 +167,15 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -191,12 +186,12 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_invoke_simple_request(grpc_end2end_test_config config) { - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_simple_request", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_invoke_simple_request", NULL, NULL); void *rc = grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234"); @@ -207,7 +202,8 @@ static void test_invoke_simple_request(grpc_end2end_test_config config) { static void test_invoke_10_simple_requests(grpc_end2end_test_config config) { int i; - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_10_simple_requests", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_invoke_10_simple_requests", NULL, NULL); void *rc = grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234"); diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c index b1c6c98b0f..8b8a11babe 100644 --- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Request/response with metadata and payload.*/ @@ -128,9 +127,9 @@ static void test_request_response_with_metadata_and_payload( "\xf0\xf1\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xfb\xfc\xfd\xfe\xff", 16, {{NULL, NULL, NULL}}}}; - grpc_end2end_test_fixture f = begin_test(config, "test_request_response_with_metadata_and_payload", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = begin_test( + config, "test_request_response_with_metadata_and_payload", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -144,8 +143,8 @@ static void test_request_response_with_metadata_and_payload( size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -183,12 +182,11 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -202,8 +200,8 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -222,11 +220,9 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -257,8 +253,7 @@ static void test_request_response_with_metadata_and_payload( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c index d0a7b9a4b5..ef6dfe9561 100644 --- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Request/response with metadata and payload.*/ @@ -114,9 +113,9 @@ static void test_request_response_with_metadata_and_payload( {"key2", "val2", 4, {{NULL, NULL, NULL}}}}; grpc_metadata meta_s[2] = {{"key3", "val3", 4, {{NULL, NULL, NULL}}}, {"key4", "val4", 4, {{NULL, NULL, NULL}}}}; - grpc_end2end_test_fixture f = begin_test(config, "test_request_response_with_metadata_and_payload", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = begin_test( + config, "test_request_response_with_metadata_and_payload", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -130,8 +129,8 @@ static void test_request_response_with_metadata_and_payload( size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -169,12 +168,11 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -188,8 +186,8 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -208,11 +206,9 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -235,8 +231,7 @@ static void test_request_response_with_metadata_and_payload( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c index 9bf4816b4e..38d3432f94 100644 --- a/test/core/end2end/tests/request_response_with_payload.c +++ b/test/core/end2end/tests/request_response_with_payload.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void request_response_with_payload(grpc_end2end_test_fixture f) { @@ -108,8 +107,7 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { grpc_byte_buffer *response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1); gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -123,8 +121,8 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -161,12 +159,11 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -179,8 +176,8 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -199,11 +196,9 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -222,8 +217,7 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); @@ -235,7 +229,8 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { payload and status. */ static void test_invoke_request_response_with_payload( grpc_end2end_test_config config) { - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_request_response_with_payload", NULL, NULL); + grpc_end2end_test_fixture f = begin_test( + config, "test_invoke_request_response_with_payload", NULL, NULL); request_response_with_payload(f); end_test(&f); config.tear_down_data(&f); @@ -244,7 +239,8 @@ static void test_invoke_request_response_with_payload( static void test_invoke_10_request_response_with_payload( grpc_end2end_test_config config) { int i; - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_10_request_response_with_payload", NULL, NULL); + grpc_end2end_test_fixture f = begin_test( + config, "test_invoke_10_request_response_with_payload", NULL, NULL); for (i = 0; i < 10; i++) { request_response_with_payload(f); } diff --git a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c index 3de1d86757..665ad3d2a6 100644 --- a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c +++ b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c @@ -51,11 +51,7 @@ static const char iam_selector[] = "selector"; static const char overridden_iam_token[] = "overridden_token"; static const char overridden_iam_selector[] = "overridden_selector"; -typedef enum { - NONE, - OVERRIDE, - DESTROY -} override_mode; +typedef enum { NONE, OVERRIDE, DESTROY } override_mode; enum { TIMEOUT = 200000 }; @@ -88,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -104,12 +102,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void print_auth_context(int is_client, const grpc_auth_context *ctx) { @@ -132,10 +127,11 @@ static void print_auth_context(int is_client, const grpc_auth_context *ctx) { static void test_call_creds_failure(grpc_end2end_test_config config) { grpc_call *c; grpc_credentials *creds = NULL; - grpc_end2end_test_fixture f = begin_test(config, "test_call_creds_failure", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_call_creds_failure", NULL, NULL); gpr_timespec deadline = five_seconds_time(); - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); /* Try with credentials unfit to be set on a call (channel creds). */ @@ -163,8 +159,7 @@ static void request_response_with_payload_and_call_creds( gpr_timespec deadline = five_seconds_time(); grpc_end2end_test_fixture f = begin_test(config, test_name, NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -180,8 +175,8 @@ static void request_response_with_payload_and_call_creds( grpc_credentials *creds = NULL; const grpc_auth_context *s_auth_context = NULL; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); creds = grpc_iam_credentials_create(iam_token, iam_selector); GPR_ASSERT(creds != NULL); @@ -236,13 +231,11 @@ static void request_response_with_payload_and_call_creds( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, - &call_details, - &request_metadata_recv, - f.server_cq, f.server_cq, - tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); s_auth_context = grpc_call_auth_context(s); GPR_ASSERT(s_auth_context != NULL); print_auth_context(0, s_auth_context); @@ -261,8 +254,8 @@ static void request_response_with_payload_and_call_creds( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -281,11 +274,9 @@ static void request_response_with_payload_and_call_creds( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -337,8 +328,7 @@ static void request_response_with_payload_and_call_creds( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); @@ -351,17 +341,22 @@ static void request_response_with_payload_and_call_creds( void test_request_response_with_payload_and_call_creds( grpc_end2end_test_config config) { - request_response_with_payload_and_call_creds("test_request_response_with_payload_and_call_creds", config, NONE); + request_response_with_payload_and_call_creds( + "test_request_response_with_payload_and_call_creds", config, NONE); } void test_request_response_with_payload_and_overridden_call_creds( grpc_end2end_test_config config) { - request_response_with_payload_and_call_creds("test_request_response_with_payload_and_overridden_call_creds", config, OVERRIDE); + request_response_with_payload_and_call_creds( + "test_request_response_with_payload_and_overridden_call_creds", config, + OVERRIDE); } void test_request_response_with_payload_and_deleted_call_creds( grpc_end2end_test_config config) { - request_response_with_payload_and_call_creds("test_request_response_with_payload_and_deleted_call_creds", config, DESTROY); + request_response_with_payload_and_call_creds( + "test_request_response_with_payload_and_deleted_call_creds", config, + DESTROY); } void grpc_end2end_tests(grpc_end2end_test_config config) { @@ -372,4 +367,3 @@ void grpc_end2end_tests(grpc_end2end_test_config config) { test_request_response_with_payload_and_deleted_call_creds(config); } } - diff --git a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c index e91ebf41c6..a5c0851d05 100644 --- a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Request/response with metadata and payload.*/ @@ -110,12 +109,15 @@ static void test_request_response_with_metadata_and_payload( grpc_byte_buffer *response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1); gpr_timespec deadline = five_seconds_time(); - grpc_metadata meta_c[2] = {{"key1", "val1", 4, {{NULL, NULL, NULL}}}, {"key2", "val2", 4, {{NULL, NULL, NULL}}}}; - grpc_metadata meta_s[2] = {{"key3", "val3", 4, {{NULL, NULL, NULL}}}, {"key4", "val4", 4, {{NULL, NULL, NULL}}}}; - grpc_metadata meta_t[2] = {{"key5", "val5", 4, {{NULL, NULL, NULL}}}, {"key6", "val6", 4, {{NULL, NULL, NULL}}}}; - grpc_end2end_test_fixture f = begin_test(config, "test_request_response_with_metadata_and_payload", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_metadata meta_c[2] = {{"key1", "val1", 4, {{NULL, NULL, NULL}}}, + {"key2", "val2", 4, {{NULL, NULL, NULL}}}}; + grpc_metadata meta_s[2] = {{"key3", "val3", 4, {{NULL, NULL, NULL}}}, + {"key4", "val4", 4, {{NULL, NULL, NULL}}}}; + grpc_metadata meta_t[2] = {{"key5", "val5", 4, {{NULL, NULL, NULL}}}, + {"key6", "val6", 4, {{NULL, NULL, NULL}}}}; + grpc_end2end_test_fixture f = begin_test( + config, "test_request_response_with_metadata_and_payload", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -129,8 +131,8 @@ static void test_request_response_with_metadata_and_payload( size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -168,13 +170,11 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, - &call_details, - &request_metadata_recv, - f.server_cq, f.server_cq, - tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -188,8 +188,8 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -209,11 +209,9 @@ static void test_request_response_with_metadata_and_payload( op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -237,8 +235,7 @@ static void test_request_response_with_metadata_and_payload( grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(response_payload); diff --git a/test/core/end2end/tests/request_with_flags.c b/test/core/end2end/tests/request_with_flags.c index 5ad475544c..fac0602328 100644 --- a/test/core/end2end/tests/request_with_flags.c +++ b/test/core/end2end/tests/request_with_flags.c @@ -75,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -91,12 +93,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void test_invoke_request_with_flags( @@ -109,8 +108,7 @@ static void test_invoke_request_with_flags( gpr_timespec deadline = five_seconds_time(); grpc_end2end_test_fixture f = begin_test(config, "test_invoke_request_with_flags", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -123,8 +121,8 @@ static void test_invoke_request_with_flags( size_t details_capacity = 0; grpc_call_error expectation; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -166,8 +164,7 @@ static void test_invoke_request_with_flags( grpc_call_destroy(c); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(request_payload_recv); @@ -178,7 +175,7 @@ static void test_invoke_request_with_flags( void grpc_end2end_tests(grpc_end2end_test_config config) { size_t i; - gpr_uint32 flags_for_op[GRPC_OP_RECV_CLOSE_ON_SERVER+1]; + gpr_uint32 flags_for_op[GRPC_OP_RECV_CLOSE_ON_SERVER + 1]; { /* check that all grpc_op_types fail when their flag value is set to an diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c index cf6f4b23e4..ad34c69774 100644 --- a/test/core/end2end/tests/request_with_large_metadata.c +++ b/test/core/end2end/tests/request_with_large_metadata.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Request with a large amount of metadata.*/ @@ -107,9 +106,9 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { grpc_raw_byte_buffer_create(&request_payload_slice, 1); gpr_timespec deadline = five_seconds_time(); grpc_metadata meta; - grpc_end2end_test_fixture f = begin_test(config, "test_request_with_large_metadata", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = + begin_test(config, "test_request_with_large_metadata", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -123,8 +122,8 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { int was_cancelled = 2; const int large_size = 64 * 1024; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); meta.key = "key"; @@ -164,12 +163,11 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -182,8 +180,8 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -198,11 +196,9 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -221,8 +217,7 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(request_payload_recv); diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c index 935aa5878b..8db6457830 100644 --- a/test/core/end2end/tests/request_with_payload.c +++ b/test/core/end2end/tests/request_with_payload.c @@ -74,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -90,12 +92,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } /* Client sends a request with payload, server reads then returns status. */ @@ -106,9 +105,9 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { grpc_byte_buffer *request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); gpr_timespec deadline = five_seconds_time(); - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_request_with_payload", NULL, NULL); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + grpc_end2end_test_fixture f = + begin_test(config, "test_invoke_request_with_payload", NULL, NULL); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -121,8 +120,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -155,12 +154,11 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -173,8 +171,8 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); + cq_expect_completion(cqv, tag(102), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; @@ -189,11 +187,9 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); - cq_expect_completion(v_server, tag(103), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(103), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -211,8 +207,7 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); grpc_byte_buffer_destroy(request_payload); grpc_byte_buffer_destroy(request_payload_recv); diff --git a/test/core/end2end/tests/server_finishes_request.c b/test/core/end2end/tests/server_finishes_request.c index 973f024b4d..062a59aca2 100644 --- a/test/core/end2end/tests/server_finishes_request.c +++ b/test/core/end2end/tests/server_finishes_request.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,20 +94,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -117,7 +115,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -144,12 +142,11 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -168,17 +165,15 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -189,8 +184,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_invoke_simple_request(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c index 36558eca1d..a2665d7564 100644 --- a/test/core/end2end/tests/simple_delayed_request.c +++ b/test/core/end2end/tests/simple_delayed_request.c @@ -62,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -78,12 +80,9 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_delayed_request_body(grpc_end2end_test_config config, @@ -94,8 +93,7 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f->client_cq); - cq_verifier *v_server = cq_verifier_create(f->server_cq); + cq_verifier *cqv = cq_verifier_create(f->cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -109,8 +107,8 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, config.init_client(f, client_args); - c = grpc_channel_create_call(f->client, f->client_cq, "/foo", - "foo.test.google.fr", deadline); + c = grpc_channel_create_call(f->client, f->cq, "/foo", "foo.test.google.fr", + deadline); GPR_ASSERT(c); grpc_metadata_array_init(&initial_metadata_recv); @@ -141,12 +139,12 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, config.init_server(f, server_args); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f->server, &s, &call_details, - &request_metadata_recv, f->server_cq, - f->server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f->server, &s, + &call_details, + &request_metadata_recv, + f->cq, f->cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -165,17 +163,15 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -186,8 +182,7 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_simple_delayed_request_short(grpc_end2end_test_config config) { diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c index 06f922351e..6194b841d8 100644 --- a/test/core/end2end/tests/simple_request.c +++ b/test/core/end2end/tests/simple_request.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,20 +94,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -117,7 +115,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -147,12 +145,11 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -171,17 +168,15 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -192,8 +187,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } static void test_invoke_simple_request(grpc_end2end_test_config config) { @@ -207,7 +201,8 @@ static void test_invoke_simple_request(grpc_end2end_test_config config) { static void test_invoke_10_simple_requests(grpc_end2end_test_config config) { int i; - grpc_end2end_test_fixture f = begin_test(config, "test_invoke_10_simple_requests", NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, "test_invoke_10_simple_requests", NULL, NULL); for (i = 0; i < 10; i++) { simple_request_body(f); gpr_log(GPR_INFO, "Passed simple request %d", i); @@ -217,6 +212,9 @@ static void test_invoke_10_simple_requests(grpc_end2end_test_config config) { } void grpc_end2end_tests(grpc_end2end_test_config config) { - test_invoke_simple_request(config); + int i; + for (i = 0; i < 10; i++) { + test_invoke_simple_request(config); + } test_invoke_10_simple_requests(config); } diff --git a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c index 1d7a150d7e..2cd638cbb9 100644 --- a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c +++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c @@ -76,8 +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->server_cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->server_cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + 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)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->server); f->server = NULL; } @@ -92,20 +94,16 @@ static void end_test(grpc_end2end_test_fixture *f) { shutdown_server(f); shutdown_client(f); - grpc_completion_queue_shutdown(f->server_cq); - drain_cq(f->server_cq); - grpc_completion_queue_destroy(f->server_cq); - grpc_completion_queue_shutdown(f->client_cq); - drain_cq(f->client_cq); - grpc_completion_queue_destroy(f->client_cq); + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); } static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call *c; grpc_call *s; gpr_timespec deadline = five_seconds_time(); - cq_verifier *v_client = cq_verifier_create(f.client_cq); - cq_verifier *v_server = cq_verifier_create(f.server_cq); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -117,7 +115,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { size_t details_capacity = 0; int was_cancelled = 2; - c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr:1234", deadline); GPR_ASSERT(c); @@ -147,12 +145,11 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f.server, &s, &call_details, - &request_metadata_recv, f.server_cq, - f.server_cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + cq_expect_completion(cqv, tag(101), 1); + cq_verify(cqv); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -171,17 +168,15 @@ static void simple_request_body(grpc_end2end_test_fixture f) { op++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); - cq_expect_completion(v_server, tag(102), 1); - cq_verify(v_server); - - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + cq_expect_completion(cqv, tag(102), 1); + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); - GPR_ASSERT(was_cancelled == 0); + GPR_ASSERT(was_cancelled == 1); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -192,11 +187,11 @@ static void simple_request_body(grpc_end2end_test_fixture f) { grpc_call_destroy(c); grpc_call_destroy(s); - cq_verifier_destroy(v_client); - cq_verifier_destroy(v_server); + cq_verifier_destroy(cqv); } -static void test_invoke_10_simple_requests(grpc_end2end_test_config config, int initial_sequence_number) { +static void test_invoke_10_simple_requests(grpc_end2end_test_config config, + int initial_sequence_number) { int i; grpc_end2end_test_fixture f; grpc_arg client_arg; diff --git a/test/core/fling/server.c b/test/core/fling/server.c index e58d721c86..9542e15ad0 100644 --- a/test/core/fling/server.c +++ b/test/core/fling/server.c @@ -234,7 +234,9 @@ int main(int argc, char **argv) { if (got_sigint && !shutdown_started) { gpr_log(GPR_INFO, "Shutting down due to SIGINT"); grpc_server_shutdown_and_notify(server, cq, tag(1000)); - GPR_ASSERT(grpc_completion_queue_pluck(cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).type == GRPC_OP_COMPLETE); + GPR_ASSERT(grpc_completion_queue_pluck( + cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)) + .type == GRPC_OP_COMPLETE); grpc_completion_queue_shutdown(cq); shutdown_started = 1; } diff --git a/test/core/httpcli/httpcli_test.c b/test/core/httpcli/httpcli_test.c index 76820916a1..6e579bc045 100644 --- a/test/core/httpcli/httpcli_test.c +++ b/test/core/httpcli/httpcli_test.c @@ -35,6 +35,7 @@ #include <string.h> +#include <grpc/grpc.h> #include "src/core/iomgr/iomgr.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> @@ -44,14 +45,16 @@ #include "test/core/util/port.h" #include "test/core/util/test_config.h" -static gpr_event g_done; +static int g_done = 0; +static grpc_httpcli_context g_context; +static grpc_pollset g_pollset; static gpr_timespec n_seconds_time(int seconds) { return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(seconds); } static void on_finish(void *arg, const grpc_httpcli_response *response) { - const char *expect = + const char *expect = "<html><head><title>Hello world!</title></head>" "<body><p>This is a test</p></body></html>"; GPR_ASSERT(arg == (void *)42); @@ -59,54 +62,69 @@ static void on_finish(void *arg, const grpc_httpcli_response *response) { GPR_ASSERT(response->status == 200); GPR_ASSERT(response->body_length == strlen(expect)); GPR_ASSERT(0 == memcmp(expect, response->body, response->body_length)); - gpr_event_set(&g_done, (void *)1); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + g_done = 1; + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } static void test_get(int use_ssl, int port) { grpc_httpcli_request req; - char* host; + char *host; + g_done = 0; gpr_log(GPR_INFO, "running %s with use_ssl=%d.", "test_get", use_ssl); gpr_asprintf(&host, "localhost:%d", port); gpr_log(GPR_INFO, "requesting from %s", host); - gpr_event_init(&g_done); memset(&req, 0, sizeof(req)); req.host = host; req.path = "/get"; req.use_ssl = use_ssl; - grpc_httpcli_get(&req, n_seconds_time(15), on_finish, (void *)42); + grpc_httpcli_get(&g_context, &g_pollset, &req, n_seconds_time(15), on_finish, + (void *)42); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (!g_done) { + grpc_pollset_work(&g_pollset, n_seconds_time(20)); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); gpr_free(host); - GPR_ASSERT(gpr_event_wait(&g_done, n_seconds_time(20))); } static void test_post(int use_ssl, int port) { grpc_httpcli_request req; - char* host; + char *host; + g_done = 0; gpr_log(GPR_INFO, "running %s with use_ssl=%d.", "test_post", (int)use_ssl); gpr_asprintf(&host, "localhost:%d", port); gpr_log(GPR_INFO, "posting to %s", host); - gpr_event_init(&g_done); memset(&req, 0, sizeof(req)); req.host = host; req.path = "/post"; req.use_ssl = use_ssl; - grpc_httpcli_post(&req, "hello", 5, n_seconds_time(15), on_finish, - (void *)42); - GPR_ASSERT(gpr_event_wait(&g_done, n_seconds_time(20))); + grpc_httpcli_post(&g_context, &g_pollset, &req, "hello", 5, + n_seconds_time(15), on_finish, (void *)42); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (!g_done) { + grpc_pollset_work(&g_pollset, n_seconds_time(20)); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); + gpr_free(host); } +static void destroy_pollset(void *ignored) { grpc_pollset_destroy(&g_pollset); } + int main(int argc, char **argv) { - gpr_subprocess* server; + gpr_subprocess *server; char *me = argv[0]; char *lslash = strrchr(me, '/'); - char* args[4]; + char *args[4]; char root[1024]; int port = grpc_pick_unused_port_or_die(); @@ -122,7 +140,7 @@ int main(int argc, char **argv) { gpr_asprintf(&args[0], "%s/../../test/core/httpcli/test_server.py", root); args[1] = "--port"; gpr_asprintf(&args[2], "%d", port); - server = gpr_subprocess_create(3, (const char**)args); + server = gpr_subprocess_create(3, (const char **)args); GPR_ASSERT(server); gpr_free(args[0]); gpr_free(args[2]); @@ -130,12 +148,16 @@ int main(int argc, char **argv) { gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(5))); grpc_test_init(argc, argv); - grpc_iomgr_init(); + grpc_init(); + grpc_httpcli_context_init(&g_context); + grpc_pollset_init(&g_pollset); test_get(0, port); test_post(0, port); - grpc_iomgr_shutdown(); + grpc_httpcli_context_destroy(&g_context); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, NULL); + grpc_shutdown(); gpr_subprocess_destroy(server); diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c index f9c5282f19..8198c24752 100644 --- a/test/core/iomgr/endpoint_tests.c +++ b/test/core/iomgr/endpoint_tests.c @@ -57,6 +57,8 @@ */ +static grpc_pollset *g_pollset; + size_t count_and_unref_slices(gpr_slice *slices, size_t nslices, int *current_data) { size_t num_bytes = 0; @@ -111,8 +113,6 @@ static gpr_slice *allocate_blocks(size_t num_bytes, size_t slice_size, struct read_and_write_test_state { grpc_endpoint *read_ep; grpc_endpoint *write_ep; - gpr_mu mu; - gpr_cv cv; size_t target_bytes; size_t bytes_read; size_t current_write_size; @@ -130,10 +130,10 @@ static void read_and_write_test_read_handler(void *data, gpr_slice *slices, GPR_ASSERT(error != GRPC_ENDPOINT_CB_ERROR); if (error == GRPC_ENDPOINT_CB_SHUTDOWN) { gpr_log(GPR_INFO, "Read handler shutdown"); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); state->read_done = 1; - gpr_cv_signal(&state->cv); - gpr_mu_unlock(&state->mu); + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); return; } @@ -141,10 +141,10 @@ static void read_and_write_test_read_handler(void *data, gpr_slice *slices, count_and_unref_slices(slices, nslices, &state->current_read_data); if (state->bytes_read == state->target_bytes) { gpr_log(GPR_INFO, "Read handler done"); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); state->read_done = 1; - gpr_cv_signal(&state->cv); - gpr_mu_unlock(&state->mu); + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); } else { grpc_endpoint_notify_on_read(state->read_ep, read_and_write_test_read_handler, data); @@ -160,14 +160,15 @@ static void read_and_write_test_write_handler(void *data, GPR_ASSERT(error != GRPC_ENDPOINT_CB_ERROR); - gpr_log(GPR_DEBUG, "%s: error=%d", "read_and_write_test_write_handler", error); + gpr_log(GPR_DEBUG, "%s: error=%d", "read_and_write_test_write_handler", + error); if (error == GRPC_ENDPOINT_CB_SHUTDOWN) { gpr_log(GPR_INFO, "Write handler shutdown"); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); state->write_done = 1; - gpr_cv_signal(&state->cv); - gpr_mu_unlock(&state->mu); + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); return; } @@ -198,10 +199,10 @@ static void read_and_write_test_write_handler(void *data, GPR_ASSERT(state->bytes_written == state->target_bytes); gpr_log(GPR_INFO, "Write handler done"); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); state->write_done = 1; - gpr_cv_signal(&state->cv); - gpr_mu_unlock(&state->mu); + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); } /* Do both reading and writing using the grpc_endpoint API. @@ -213,7 +214,8 @@ static void read_and_write_test(grpc_endpoint_test_config config, size_t slice_size, int shutdown) { struct read_and_write_test_state state; gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20); - grpc_endpoint_test_fixture f = begin_test(config, "read_and_write_test", slice_size); + grpc_endpoint_test_fixture f = + begin_test(config, "read_and_write_test", slice_size); if (shutdown) { gpr_log(GPR_INFO, "Start read and write shutdown test"); @@ -222,9 +224,6 @@ static void read_and_write_test(grpc_endpoint_test_config config, num_bytes, slice_size); } - gpr_mu_init(&state.mu); - gpr_cv_init(&state.cv); - state.read_ep = f.client_ep; state.write_ep = f.server_ep; state.target_bytes = num_bytes; @@ -253,29 +252,24 @@ static void read_and_write_test(grpc_endpoint_test_config config, grpc_endpoint_shutdown(state.write_ep); } - gpr_mu_lock(&state.mu); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); while (!state.read_done || !state.write_done) { - if (gpr_cv_wait(&state.cv, &state.mu, deadline)) { - gpr_log(GPR_ERROR, "timeout: read_done=%d, write_done=%d", - state.read_done, state.write_done); - abort(); - } + GPR_ASSERT(gpr_time_cmp(gpr_now(), deadline) < 0); + grpc_pollset_work(g_pollset, deadline); } - gpr_mu_unlock(&state.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); grpc_endpoint_destroy(state.read_ep); grpc_endpoint_destroy(state.write_ep); - gpr_mu_destroy(&state.mu); - gpr_cv_destroy(&state.cv); end_test(config); } struct timeout_test_state { - gpr_event io_done; + int io_done; }; typedef struct { - gpr_event ev; + int done; grpc_endpoint *ep; } shutdown_during_write_test_state; @@ -291,7 +285,10 @@ static void shutdown_during_write_test_read_handler( if (error != GRPC_ENDPOINT_CB_OK) { grpc_endpoint_destroy(st->ep); - gpr_event_set(&st->ev, (void *)(gpr_intptr) error); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); + st->done = error; + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); } else { grpc_endpoint_notify_on_read( st->ep, shutdown_during_write_test_read_handler, user_data); @@ -310,7 +307,10 @@ static void shutdown_during_write_test_write_handler( gpr_log(GPR_ERROR, "shutdown_during_write_test_write_handler completed unexpectedly"); } - gpr_event_set(&st->ev, (void *)(gpr_intptr) 1); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); + st->done = 1; + grpc_pollset_kick(g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); } static void shutdown_during_write_test(grpc_endpoint_test_config config, @@ -323,14 +323,15 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config, shutdown_during_write_test_state read_st; shutdown_during_write_test_state write_st; gpr_slice *slices; - grpc_endpoint_test_fixture f = begin_test(config, "shutdown_during_write_test", slice_size); + grpc_endpoint_test_fixture f = + begin_test(config, "shutdown_during_write_test", slice_size); gpr_log(GPR_INFO, "testing shutdown during a write"); read_st.ep = f.client_ep; write_st.ep = f.server_ep; - gpr_event_init(&read_st.ev); - gpr_event_init(&write_st.ev); + read_st.done = 0; + write_st.done = 0; grpc_endpoint_notify_on_read( read_st.ep, shutdown_during_write_test_read_handler, &read_st); @@ -347,9 +348,19 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config, case GRPC_ENDPOINT_WRITE_PENDING: grpc_endpoint_shutdown(write_st.ep); deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10); - GPR_ASSERT(gpr_event_wait(&write_st.ev, deadline)); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); + while (!write_st.done) { + GPR_ASSERT(gpr_time_cmp(gpr_now(), deadline) < 0); + grpc_pollset_work(g_pollset, deadline); + } + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); grpc_endpoint_destroy(write_st.ep); - GPR_ASSERT(gpr_event_wait(&read_st.ev, deadline)); + gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); + while (!read_st.done) { + GPR_ASSERT(gpr_time_cmp(gpr_now(), deadline) < 0); + grpc_pollset_work(g_pollset, deadline); + } + gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); gpr_free(slices); end_test(config); return; @@ -361,9 +372,12 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config, abort(); } -void grpc_endpoint_tests(grpc_endpoint_test_config config) { +void grpc_endpoint_tests(grpc_endpoint_test_config config, + grpc_pollset *pollset) { + g_pollset = pollset; read_and_write_test(config, 10000000, 100000, 8192, 0); read_and_write_test(config, 1000000, 100000, 1, 0); read_and_write_test(config, 100000000, 100000, 1, 1); shutdown_during_write_test(config, 1000); + g_pollset = NULL; } diff --git a/test/core/iomgr/endpoint_tests.h b/test/core/iomgr/endpoint_tests.h index 1679d7bd4f..700f854891 100644 --- a/test/core/iomgr/endpoint_tests.h +++ b/test/core/iomgr/endpoint_tests.h @@ -52,6 +52,7 @@ struct grpc_endpoint_test_config { void (*clean_up)(); }; -void grpc_endpoint_tests(grpc_endpoint_test_config config); +void grpc_endpoint_tests(grpc_endpoint_test_config config, + grpc_pollset *pollset); -#endif /* GRPC_TEST_CORE_IOMGR_ENDPOINT_TESTS_H */ +#endif /* GRPC_TEST_CORE_IOMGR_ENDPOINT_TESTS_H */ diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c index 2c8a89e4cd..fe08ec495f 100644 --- a/test/core/iomgr/fd_posix_test.c +++ b/test/core/iomgr/fd_posix_test.c @@ -51,6 +51,8 @@ #include <grpc/support/time.h> #include "test/core/util/test_config.h" +static grpc_pollset g_pollset; + /* buffer size used to send and receive data. 1024 is the minimal value to set TCP send and receive buffer. */ #define BUF_SIZE 1024 @@ -94,16 +96,12 @@ void no_op_cb(void *arg, int success) {} typedef struct { grpc_fd *em_fd; /* listening fd */ ssize_t read_bytes_total; /* total number of received bytes */ - gpr_mu mu; /* protect done and done_cv */ - gpr_cv done_cv; /* signaled when a server finishes serving */ int done; /* set to 1 when a server finishes serving */ grpc_iomgr_closure listen_closure; } server; static void server_init(server *sv) { sv->read_bytes_total = 0; - gpr_mu_init(&sv->mu); - gpr_cv_init(&sv->done_cv); sv->done = 0; } @@ -122,7 +120,7 @@ static void session_shutdown_cb(void *arg, /*session*/ int success) { session *se = arg; server *sv = se->sv; - grpc_fd_orphan(se->em_fd, NULL, NULL); + grpc_fd_orphan(se->em_fd, NULL, "a"); gpr_free(se); /* Start to shutdown listen fd. */ grpc_fd_shutdown(sv->em_fd); @@ -177,12 +175,12 @@ static void session_read_cb(void *arg, /*session*/ static void listen_shutdown_cb(void *arg /*server*/, int success) { server *sv = arg; - grpc_fd_orphan(sv->em_fd, NULL, NULL); + grpc_fd_orphan(sv->em_fd, NULL, "b"); - gpr_mu_lock(&sv->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); sv->done = 1; - gpr_cv_signal(&sv->done_cv); - gpr_mu_unlock(&sv->mu); + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } /* Called when a new TCP connection request arrives in the listening port. */ @@ -209,6 +207,7 @@ static void listen_cb(void *arg, /*=sv_arg*/ se = gpr_malloc(sizeof(*se)); se->sv = sv; se->em_fd = grpc_fd_create(fd, "listener"); + grpc_pollset_add_fd(&g_pollset, se->em_fd); se->session_read_closure.cb = session_read_cb; se->session_read_closure.cb_arg = se; grpc_fd_notify_on_read(se->em_fd, &se->session_read_closure); @@ -237,6 +236,7 @@ static int server_start(server *sv) { GPR_ASSERT(listen(fd, MAX_NUM_FD) == 0); sv->em_fd = grpc_fd_create(fd, "server"); + grpc_pollset_add_fd(&g_pollset, sv->em_fd); /* Register to be interested in reading from listen_fd. */ sv->listen_closure.cb = listen_cb; sv->listen_closure.cb_arg = sv; @@ -247,12 +247,11 @@ static int server_start(server *sv) { /* Wait and shutdown a sever. */ static void server_wait_and_shutdown(server *sv) { - gpr_mu_lock(&sv->mu); - while (!sv->done) gpr_cv_wait(&sv->done_cv, &sv->mu, gpr_inf_future); - gpr_mu_unlock(&sv->mu); - - gpr_mu_destroy(&sv->mu); - gpr_cv_destroy(&sv->done_cv); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (!sv->done) { + grpc_pollset_work(&g_pollset, gpr_inf_future); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } /* ===An upload client to test notify_on_write=== */ @@ -271,9 +270,7 @@ typedef struct { notify_on_write to schedule another write. */ int client_write_cnt; - gpr_mu mu; /* protect done and done_cv */ - gpr_cv done_cv; /* signaled when a client finishes sending */ - int done; /* set to 1 when a client finishes sending */ + int done; /* set to 1 when a client finishes sending */ grpc_iomgr_closure write_closure; } client; @@ -281,17 +278,15 @@ static void client_init(client *cl) { memset(cl->write_buf, 0, sizeof(cl->write_buf)); cl->write_bytes_total = 0; cl->client_write_cnt = 0; - gpr_mu_init(&cl->mu); - gpr_cv_init(&cl->done_cv); cl->done = 0; } /* Called when a client upload session is ready to shutdown. */ static void client_session_shutdown_cb(void *arg /*client*/, int success) { client *cl = arg; - grpc_fd_orphan(cl->em_fd, NULL, NULL); + grpc_fd_orphan(cl->em_fd, NULL, "c"); cl->done = 1; - gpr_cv_signal(&cl->done_cv); + grpc_pollset_kick(&g_pollset); } /* Write as much as possible, then register notify_on_write. */ @@ -302,9 +297,9 @@ static void client_session_write(void *arg, /*client*/ ssize_t write_once = 0; if (!success) { - gpr_mu_lock(&cl->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); client_session_shutdown_cb(arg, 1); - gpr_mu_unlock(&cl->mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); return; } @@ -314,7 +309,7 @@ static void client_session_write(void *arg, /*client*/ } while (write_once > 0); if (errno == EAGAIN) { - gpr_mu_lock(&cl->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); if (cl->client_write_cnt < CLIENT_TOTAL_WRITE_CNT) { cl->write_closure.cb = client_session_write; cl->write_closure.cb_arg = cl; @@ -323,7 +318,7 @@ static void client_session_write(void *arg, /*client*/ } else { client_session_shutdown_cb(arg, 1); } - gpr_mu_unlock(&cl->mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } else { gpr_log(GPR_ERROR, "unknown errno %s", strerror(errno)); abort(); @@ -352,18 +347,18 @@ static void client_start(client *cl, int port) { } cl->em_fd = grpc_fd_create(fd, "client"); + grpc_pollset_add_fd(&g_pollset, cl->em_fd); client_session_write(cl, 1); } /* Wait for the signal to shutdown a client. */ static void client_wait_and_shutdown(client *cl) { - gpr_mu_lock(&cl->mu); - while (!cl->done) gpr_cv_wait(&cl->done_cv, &cl->mu, gpr_inf_future); - gpr_mu_unlock(&cl->mu); - - gpr_mu_destroy(&cl->mu); - gpr_cv_destroy(&cl->done_cv); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (!cl->done) { + grpc_pollset_work(&g_pollset, gpr_inf_future); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } /* Test grpc_fd. Start an upload server and client, upload a stream of @@ -385,38 +380,29 @@ static void test_grpc_fd(void) { } typedef struct fd_change_data { - gpr_mu mu; - gpr_cv cv; void (*cb_that_ran)(void *, int success); } fd_change_data; -void init_change_data(fd_change_data *fdc) { - gpr_mu_init(&fdc->mu); - gpr_cv_init(&fdc->cv); - fdc->cb_that_ran = NULL; -} +void init_change_data(fd_change_data *fdc) { fdc->cb_that_ran = NULL; } -void destroy_change_data(fd_change_data *fdc) { - gpr_mu_destroy(&fdc->mu); - gpr_cv_destroy(&fdc->cv); -} +void destroy_change_data(fd_change_data *fdc) {} static void first_read_callback(void *arg /* fd_change_data */, int success) { fd_change_data *fdc = arg; - gpr_mu_lock(&fdc->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); fdc->cb_that_ran = first_read_callback; - gpr_cv_signal(&fdc->cv); - gpr_mu_unlock(&fdc->mu); + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } static void second_read_callback(void *arg /* fd_change_data */, int success) { fd_change_data *fdc = arg; - gpr_mu_lock(&fdc->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); fdc->cb_that_ran = second_read_callback; - gpr_cv_signal(&fdc->cv); - gpr_mu_unlock(&fdc->mu); + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } /* Test that changing the callback we use for notify_on_read actually works. @@ -448,6 +434,7 @@ static void test_grpc_fd_change(void) { GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0); em_fd = grpc_fd_create(sv[0], "test_grpc_fd_change"); + grpc_pollset_add_fd(&g_pollset, em_fd); /* Register the first callback, then make its FD readable */ grpc_fd_notify_on_read(em_fd, &first_closure); @@ -456,12 +443,12 @@ static void test_grpc_fd_change(void) { GPR_ASSERT(result == 1); /* And now wait for it to run. */ - gpr_mu_lock(&a.mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (a.cb_that_ran == NULL) { - gpr_cv_wait(&a.cv, &a.mu, gpr_inf_future); + grpc_pollset_work(&g_pollset, gpr_inf_future); } GPR_ASSERT(a.cb_that_ran == first_read_callback); - gpr_mu_unlock(&a.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); /* And drain the socket so we can generate a new read edge */ result = read(sv[0], &data, 1); @@ -474,25 +461,29 @@ static void test_grpc_fd_change(void) { result = write(sv[1], &data, 1); GPR_ASSERT(result == 1); - gpr_mu_lock(&b.mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (b.cb_that_ran == NULL) { - gpr_cv_wait(&b.cv, &b.mu, gpr_inf_future); + grpc_pollset_work(&g_pollset, gpr_inf_future); } /* Except now we verify that second_read_callback ran instead */ GPR_ASSERT(b.cb_that_ran == second_read_callback); - gpr_mu_unlock(&b.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); - grpc_fd_orphan(em_fd, NULL, NULL); + grpc_fd_orphan(em_fd, NULL, "d"); destroy_change_data(&a); destroy_change_data(&b); close(sv[1]); } +static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } + int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_iomgr_init(); + grpc_pollset_init(&g_pollset); test_grpc_fd(); test_grpc_fd_change(); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset); grpc_iomgr_shutdown(); return 0; } diff --git a/test/core/iomgr/poll_kick_posix_test.c b/test/core/iomgr/poll_kick_posix_test.c index 2c5b444d3a..3aa6807806 100644 --- a/test/core/iomgr/poll_kick_posix_test.c +++ b/test/core/iomgr/poll_kick_posix_test.c @@ -31,7 +31,7 @@ * */ -#include "src/core/iomgr/pollset_kick.h" +#include "src/core/iomgr/pollset_kick_posix.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> @@ -45,31 +45,31 @@ static void test_allocation(void) { static void test_non_kick(void) { grpc_pollset_kick_state state; - int fd; + grpc_kick_fd_info *kfd; grpc_pollset_kick_init(&state); - fd = grpc_pollset_kick_pre_poll(&state); - GPR_ASSERT(fd >= 0); + kfd = grpc_pollset_kick_pre_poll(&state); + GPR_ASSERT(kfd != NULL); - grpc_pollset_kick_post_poll(&state); + grpc_pollset_kick_post_poll(&state, kfd); grpc_pollset_kick_destroy(&state); } static void test_basic_kick(void) { /* Kicked during poll */ grpc_pollset_kick_state state; - int fd; + grpc_kick_fd_info *kfd; grpc_pollset_kick_init(&state); - fd = grpc_pollset_kick_pre_poll(&state); - GPR_ASSERT(fd >= 0); + kfd = grpc_pollset_kick_pre_poll(&state); + GPR_ASSERT(kfd != NULL); grpc_pollset_kick_kick(&state); /* Now hypothetically we polled and found that we were kicked */ - grpc_pollset_kick_consume(&state); + grpc_pollset_kick_consume(&state, kfd); - grpc_pollset_kick_post_poll(&state); + grpc_pollset_kick_post_poll(&state, kfd); grpc_pollset_kick_destroy(&state); } @@ -77,13 +77,13 @@ static void test_basic_kick(void) { static void test_non_poll_kick(void) { /* Kick before entering poll */ grpc_pollset_kick_state state; - int fd; + grpc_kick_fd_info *kfd; grpc_pollset_kick_init(&state); grpc_pollset_kick_kick(&state); - fd = grpc_pollset_kick_pre_poll(&state); - GPR_ASSERT(fd < 0); + kfd = grpc_pollset_kick_pre_poll(&state); + GPR_ASSERT(kfd == NULL); grpc_pollset_kick_destroy(&state); } @@ -92,20 +92,20 @@ static void test_non_poll_kick(void) { static void test_over_free(void) { /* Check high watermark pipe free logic */ int i; - struct grpc_pollset_kick_state *kick_state = - gpr_malloc(sizeof(grpc_pollset_kick_state) * GRPC_MAX_CACHED_PIPES); + grpc_kick_fd_info **kfds = + gpr_malloc(sizeof(grpc_kick_fd_info *) * GRPC_MAX_CACHED_PIPES); + grpc_pollset_kick_state state; + grpc_pollset_kick_init(&state); for (i = 0; i < GRPC_MAX_CACHED_PIPES; ++i) { - int fd; - grpc_pollset_kick_init(&kick_state[i]); - fd = grpc_pollset_kick_pre_poll(&kick_state[i]); - GPR_ASSERT(fd >= 0); + kfds[i] = grpc_pollset_kick_pre_poll(&state); + GPR_ASSERT(kfds[i] != NULL); } for (i = 0; i < GRPC_MAX_CACHED_PIPES; ++i) { - grpc_pollset_kick_post_poll(&kick_state[i]); - grpc_pollset_kick_destroy(&kick_state[i]); + grpc_pollset_kick_post_poll(&state, kfds[i]); } - gpr_free(kick_state); + grpc_pollset_kick_destroy(&state); + gpr_free(kfds); } static void run_tests(void) { diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c index 3c4d8fed4f..b673c032b2 100644 --- a/test/core/iomgr/tcp_client_posix_test.c +++ b/test/core/iomgr/tcp_client_posix_test.c @@ -45,20 +45,31 @@ #include <grpc/support/time.h> #include "test/core/util/test_config.h" +static grpc_pollset_set g_pollset_set; +static grpc_pollset g_pollset; +static int g_connections_complete = 0; + static gpr_timespec test_deadline(void) { return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10); } +static void finish_connection() { + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + g_connections_complete++; + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); +} + static void must_succeed(void *arg, grpc_endpoint *tcp) { GPR_ASSERT(tcp); grpc_endpoint_shutdown(tcp); grpc_endpoint_destroy(tcp); - gpr_event_set(arg, (void *)1); + finish_connection(); } static void must_fail(void *arg, grpc_endpoint *tcp) { GPR_ASSERT(!tcp); - gpr_event_set(arg, (void *)1); + finish_connection(); } void test_succeeds(void) { @@ -66,9 +77,7 @@ void test_succeeds(void) { socklen_t addr_len = sizeof(addr); int svr_fd; int r; - gpr_event ev; - - gpr_event_init(&ev); + int connections_complete_before; memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; @@ -79,10 +88,14 @@ void test_succeeds(void) { GPR_ASSERT(0 == bind(svr_fd, (struct sockaddr *)&addr, addr_len)); GPR_ASSERT(0 == listen(svr_fd, 1)); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + connections_complete_before = g_connections_complete; + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); + /* connect to it */ GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0); - grpc_tcp_client_connect(must_succeed, &ev, (struct sockaddr *)&addr, addr_len, - gpr_inf_future); + grpc_tcp_client_connect(must_succeed, NULL, &g_pollset_set, + (struct sockaddr *)&addr, addr_len, gpr_inf_future); /* await the connection */ do { @@ -92,26 +105,39 @@ void test_succeeds(void) { GPR_ASSERT(r >= 0); close(r); - /* wait for the connection callback to finish */ - GPR_ASSERT(gpr_event_wait(&ev, test_deadline())); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + + while (g_connections_complete == connections_complete_before) { + grpc_pollset_work(&g_pollset, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)); + } + + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } void test_fails(void) { struct sockaddr_in addr; socklen_t addr_len = sizeof(addr); - gpr_event ev; - - gpr_event_init(&ev); + int connections_complete_before; memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + connections_complete_before = g_connections_complete; + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); + /* connect to a broken address */ - grpc_tcp_client_connect(must_fail, &ev, (struct sockaddr *)&addr, addr_len, - gpr_inf_future); + grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set, + (struct sockaddr *)&addr, addr_len, gpr_inf_future); + + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); /* wait for the connection callback to finish */ - GPR_ASSERT(gpr_event_wait(&ev, test_deadline())); + while (g_connections_complete == connections_complete_before) { + grpc_pollset_work(&g_pollset, test_deadline()); + } + + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } void test_times_out(void) { @@ -122,11 +148,9 @@ void test_times_out(void) { int client_fd[NUM_CLIENT_CONNECTS]; int i; int r; - gpr_event ev; + int connections_complete_before; gpr_timespec connect_deadline; - gpr_event_init(&ev); - memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; @@ -153,28 +177,50 @@ void test_times_out(void) { connect_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1); - grpc_tcp_client_connect(must_fail, &ev, (struct sockaddr *)&addr, addr_len, - connect_deadline); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + connections_complete_before = g_connections_complete; + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); + + grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set, + (struct sockaddr *)&addr, addr_len, connect_deadline); + /* Make sure the event doesn't trigger early */ - GPR_ASSERT(!gpr_event_wait(&ev, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(500))); - /* Now wait until it should have triggered */ - sleep(1); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (gpr_time_cmp(gpr_time_add(connect_deadline, gpr_time_from_seconds(2)), + gpr_now()) > 0) { + int is_after_deadline = gpr_time_cmp(connect_deadline, gpr_now()) <= 0; + if (is_after_deadline && + gpr_time_cmp(gpr_time_add(connect_deadline, gpr_time_from_seconds(1)), + gpr_now()) > 0) { + /* allow some slack before insisting that things be done */ + } else { + GPR_ASSERT(g_connections_complete == + connections_complete_before + is_after_deadline); + } + grpc_pollset_work(&g_pollset, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(10)); + } + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); - /* wait for the connection callback to finish */ - GPR_ASSERT(gpr_event_wait(&ev, test_deadline())); close(svr_fd); for (i = 0; i < NUM_CLIENT_CONNECTS; ++i) { close(client_fd[i]); } } +static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } + int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_iomgr_init(); + grpc_pollset_set_init(&g_pollset_set); + grpc_pollset_init(&g_pollset); + grpc_pollset_set_add_pollset(&g_pollset_set, &g_pollset); test_succeeds(); gpr_log(GPR_ERROR, "End of first test"); test_fails(); test_times_out(); + grpc_pollset_set_destroy(&g_pollset_set); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset); grpc_iomgr_shutdown(); return 0; } diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c index 2cfcc8311c..a23c64928e 100644 --- a/test/core/iomgr/tcp_posix_test.c +++ b/test/core/iomgr/tcp_posix_test.c @@ -48,6 +48,8 @@ #include "test/core/util/test_config.h" #include "test/core/iomgr/endpoint_tests.h" +static grpc_pollset g_pollset; + /* General test notes: @@ -114,8 +116,6 @@ static size_t fill_socket_partial(int fd, size_t bytes) { struct read_socket_state { grpc_endpoint *ep; - gpr_mu mu; - gpr_cv cv; ssize_t read_bytes; ssize_t target_read_bytes; }; @@ -145,18 +145,18 @@ static void read_cb(void *user_data, gpr_slice *slices, size_t nslices, GPR_ASSERT(error == GRPC_ENDPOINT_CB_OK); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); current_data = state->read_bytes % 256; read_bytes = count_and_unref_slices(slices, nslices, ¤t_data); state->read_bytes += read_bytes; gpr_log(GPR_INFO, "Read %d bytes of %d", read_bytes, state->target_read_bytes); if (state->read_bytes >= state->target_read_bytes) { - gpr_cv_signal(&state->cv); + /* empty */ } else { grpc_endpoint_notify_on_read(state->ep, read_cb, state); } - gpr_mu_unlock(&state->mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } /* Write to a socket, then read from it using the grpc_tcp API. */ @@ -173,31 +173,25 @@ static void read_test(ssize_t num_bytes, ssize_t slice_size) { create_sockets(sv); ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), slice_size); + grpc_endpoint_add_to_pollset(ep, &g_pollset); + written_bytes = fill_socket_partial(sv[0], num_bytes); gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes); - gpr_mu_init(&state.mu); - gpr_cv_init(&state.cv); state.ep = ep; state.read_bytes = 0; state.target_read_bytes = written_bytes; grpc_endpoint_notify_on_read(ep, read_cb, &state); - gpr_mu_lock(&state.mu); - for (;;) { - GPR_ASSERT(gpr_cv_wait(&state.cv, &state.mu, deadline) == 0); - if (state.read_bytes >= state.target_read_bytes) { - break; - } + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (state.read_bytes < state.target_read_bytes) { + grpc_pollset_work(&g_pollset, deadline); } GPR_ASSERT(state.read_bytes == state.target_read_bytes); - gpr_mu_unlock(&state.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); grpc_endpoint_destroy(ep); - - gpr_mu_destroy(&state.mu); - gpr_cv_destroy(&state.cv); } /* Write to a socket until it fills up, then read from it using the grpc_tcp @@ -214,37 +208,29 @@ static void large_read_test(ssize_t slice_size) { create_sockets(sv); ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), slice_size); + grpc_endpoint_add_to_pollset(ep, &g_pollset); + written_bytes = fill_socket(sv[0]); gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes); - gpr_mu_init(&state.mu); - gpr_cv_init(&state.cv); state.ep = ep; state.read_bytes = 0; state.target_read_bytes = written_bytes; grpc_endpoint_notify_on_read(ep, read_cb, &state); - gpr_mu_lock(&state.mu); - for (;;) { - GPR_ASSERT(gpr_cv_wait(&state.cv, &state.mu, deadline) == 0); - if (state.read_bytes >= state.target_read_bytes) { - break; - } + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + while (state.read_bytes < state.target_read_bytes) { + grpc_pollset_work(&g_pollset, deadline); } GPR_ASSERT(state.read_bytes == state.target_read_bytes); - gpr_mu_unlock(&state.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); grpc_endpoint_destroy(ep); - - gpr_mu_destroy(&state.mu); - gpr_cv_destroy(&state.cv); } struct write_socket_state { grpc_endpoint *ep; - gpr_mu mu; - gpr_cv cv; int write_done; }; @@ -275,11 +261,11 @@ static void write_done(void *user_data /* write_socket_state */, grpc_endpoint_cb_status error) { struct write_socket_state *state = (struct write_socket_state *)user_data; gpr_log(GPR_INFO, "Write done callback called"); - gpr_mu_lock(&state->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); gpr_log(GPR_INFO, "Signalling write done"); state->write_done = 1; - gpr_cv_signal(&state->cv); - gpr_mu_unlock(&state->mu); + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) { @@ -294,6 +280,9 @@ void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) { GPR_ASSERT(fcntl(fd, F_SETFL, flags & ~O_NONBLOCK) == 0); for (;;) { + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + grpc_pollset_work(&g_pollset, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(10)); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); do { bytes_read = read(fd, buf, bytes_left > read_size ? read_size : bytes_left); @@ -352,9 +341,8 @@ static void write_test(ssize_t num_bytes, ssize_t slice_size) { ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"), GRPC_TCP_DEFAULT_READ_SLICE_SIZE); + grpc_endpoint_add_to_pollset(ep, &g_pollset); - gpr_mu_init(&state.mu); - gpr_cv_init(&state.cv); state.ep = ep; state.write_done = 0; @@ -367,19 +355,17 @@ static void write_test(ssize_t num_bytes, ssize_t slice_size) { GPR_ASSERT(read_bytes == num_bytes); } else { drain_socket_blocking(sv[0], num_bytes, num_bytes); - gpr_mu_lock(&state.mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); for (;;) { if (state.write_done) { break; } - GPR_ASSERT(gpr_cv_wait(&state.cv, &state.mu, deadline) == 0); + grpc_pollset_work(&g_pollset, deadline); } - gpr_mu_unlock(&state.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } grpc_endpoint_destroy(ep); - gpr_mu_destroy(&state.mu); - gpr_cv_destroy(&state.cv); gpr_free(slices); } @@ -409,10 +395,10 @@ static void write_error_test(ssize_t num_bytes, ssize_t slice_size) { ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_error_test"), GRPC_TCP_DEFAULT_READ_SLICE_SIZE); + grpc_endpoint_add_to_pollset(ep, &g_pollset); + close(sv[0]); - gpr_mu_init(&state.mu); - gpr_cv_init(&state.cv); state.ep = ep; state.write_done = 0; @@ -425,20 +411,18 @@ static void write_error_test(ssize_t num_bytes, ssize_t slice_size) { break; case GRPC_ENDPOINT_WRITE_PENDING: grpc_endpoint_notify_on_read(ep, read_done_for_write_error, NULL); - gpr_mu_lock(&state.mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); for (;;) { if (state.write_done) { break; } - GPR_ASSERT(gpr_cv_wait(&state.cv, &state.mu, deadline) == 0); + grpc_pollset_work(&g_pollset, deadline); } - gpr_mu_unlock(&state.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); break; } grpc_endpoint_destroy(ep); - gpr_mu_destroy(&state.mu); - gpr_cv_destroy(&state.cv); free(slices); } @@ -479,6 +463,8 @@ static grpc_endpoint_test_fixture create_fixture_tcp_socketpair( grpc_tcp_create(grpc_fd_create(sv[0], "fixture:client"), slice_size); f.server_ep = grpc_tcp_create(grpc_fd_create(sv[1], "fixture:server"), slice_size); + grpc_endpoint_add_to_pollset(f.client_ep, &g_pollset); + grpc_endpoint_add_to_pollset(f.server_ep, &g_pollset); return f; } @@ -487,11 +473,15 @@ static grpc_endpoint_test_config configs[] = { {"tcp/tcp_socketpair", create_fixture_tcp_socketpair, clean_up}, }; +static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } + int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_init(); + grpc_pollset_init(&g_pollset); run_tests(); - grpc_endpoint_tests(configs[0]); + grpc_endpoint_tests(configs[0], &g_pollset); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset); grpc_shutdown(); return 0; diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 328b19f68a..fb262711c0 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -45,18 +45,17 @@ #define LOG_TEST(x) gpr_log(GPR_INFO, "%s", #x) -static gpr_mu mu; -static gpr_cv cv; -static int nconnects = 0; +static grpc_pollset g_pollset; +static int g_nconnects = 0; static void on_connect(void *arg, grpc_endpoint *tcp) { grpc_endpoint_shutdown(tcp); grpc_endpoint_destroy(tcp); - gpr_mu_lock(&mu); - nconnects++; - gpr_cv_broadcast(&cv); - gpr_mu_unlock(&mu); + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); + g_nconnects++; + grpc_pollset_kick(&g_pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } static void test_no_op(void) { @@ -106,12 +105,11 @@ static void test_connect(int n) { grpc_tcp_server *s = grpc_tcp_server_create(); int nconnects_before; gpr_timespec deadline; + grpc_pollset *pollsets[1]; int i; LOG_TEST("test_connect"); gpr_log(GPR_INFO, "clients=%d", n); - gpr_mu_lock(&mu); - memset(&addr, 0, sizeof(addr)); addr.ss_family = AF_INET; GPR_ASSERT(grpc_tcp_server_add_port(s, (struct sockaddr *)&addr, addr_len)); @@ -121,38 +119,42 @@ static void test_connect(int n) { GPR_ASSERT(getsockname(svrfd, (struct sockaddr *)&addr, &addr_len) == 0); GPR_ASSERT(addr_len <= sizeof(addr)); - grpc_tcp_server_start(s, NULL, 0, on_connect, NULL); + pollsets[0] = &g_pollset; + grpc_tcp_server_start(s, pollsets, 1, on_connect, NULL); + + gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); for (i = 0; i < n; i++) { - deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1); + deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(4000); - nconnects_before = nconnects; + nconnects_before = g_nconnects; clifd = socket(addr.ss_family, SOCK_STREAM, 0); GPR_ASSERT(clifd >= 0); + gpr_log(GPR_DEBUG, "start connect"); GPR_ASSERT(connect(clifd, (struct sockaddr *)&addr, addr_len) == 0); - while (nconnects == nconnects_before) { - GPR_ASSERT(gpr_cv_wait(&cv, &mu, deadline) == 0); + gpr_log(GPR_DEBUG, "wait"); + while (g_nconnects == nconnects_before && + gpr_time_cmp(deadline, gpr_now()) > 0) { + grpc_pollset_work(&g_pollset, deadline); } + gpr_log(GPR_DEBUG, "wait done"); - GPR_ASSERT(nconnects == nconnects_before + 1); + GPR_ASSERT(g_nconnects == nconnects_before + 1); close(clifd); - - if (i != n - 1) { - sleep(1); - } } - gpr_mu_unlock(&mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); grpc_tcp_server_destroy(s, NULL, NULL); } +static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } + int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_iomgr_init(); - gpr_mu_init(&mu); - gpr_cv_init(&cv); + grpc_pollset_init(&g_pollset); test_no_op(); test_no_op_with_start(); @@ -161,8 +163,7 @@ int main(int argc, char **argv) { test_connect(1); test_connect(10); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset); grpc_iomgr_shutdown(); - gpr_mu_destroy(&mu); - gpr_cv_destroy(&cv); return 0; } diff --git a/test/core/security/credentials_test.c b/test/core/security/credentials_test.c index 69ec680c18..4253be6b07 100644 --- a/test/core/security/credentials_test.c +++ b/test/core/security/credentials_test.c @@ -210,8 +210,7 @@ static void test_oauth2_token_fetcher_creds_parsing_ok(void) { grpc_httpcli_response response = http_response(200, valid_oauth2_json_response); GPR_ASSERT(grpc_oauth2_token_fetcher_credentials_parse_server_response( - &response, &token_md, &token_lifetime) == - GRPC_CREDENTIALS_OK); + &response, &token_md, &token_lifetime) == GRPC_CREDENTIALS_OK); GPR_ASSERT(token_lifetime.tv_sec == 3599); GPR_ASSERT(token_lifetime.tv_nsec == 0); GPR_ASSERT(token_md->num_entries == 1); @@ -328,7 +327,7 @@ static void test_iam_creds(void) { test_iam_authorization_token, test_iam_authority_selector); GPR_ASSERT(grpc_credentials_has_request_metadata(creds)); GPR_ASSERT(grpc_credentials_has_request_metadata_only(creds)); - grpc_credentials_get_request_metadata(creds, test_service_url, + grpc_credentials_get_request_metadata(creds, NULL, test_service_url, check_iam_metadata, creds); } @@ -354,8 +353,8 @@ static void test_ssl_oauth2_composite_creds(void) { grpc_composite_credentials_create(ssl_creds, oauth2_creds); grpc_credentials_unref(ssl_creds); grpc_credentials_unref(oauth2_creds); - GPR_ASSERT(strcmp(composite_creds->type, - GRPC_CREDENTIALS_TYPE_COMPOSITE) == 0); + GPR_ASSERT(strcmp(composite_creds->type, GRPC_CREDENTIALS_TYPE_COMPOSITE) == + 0); GPR_ASSERT(grpc_credentials_has_request_metadata(composite_creds)); GPR_ASSERT(!grpc_credentials_has_request_metadata_only(composite_creds)); creds_array = grpc_composite_credentials_get_credentials(composite_creds); @@ -364,14 +363,13 @@ static void test_ssl_oauth2_composite_creds(void) { GRPC_CREDENTIALS_TYPE_SSL) == 0); GPR_ASSERT(strcmp(creds_array->creds_array[1]->type, GRPC_CREDENTIALS_TYPE_OAUTH2) == 0); - grpc_credentials_get_request_metadata(composite_creds, test_service_url, + grpc_credentials_get_request_metadata(composite_creds, NULL, test_service_url, check_ssl_oauth2_composite_metadata, composite_creds); } void test_ssl_fake_transport_security_composite_creds_failure(void) { - grpc_credentials *ssl_creds = - grpc_ssl_credentials_create(NULL, NULL); + grpc_credentials *ssl_creds = grpc_ssl_credentials_create(NULL, NULL); grpc_credentials *fake_transport_security_creds = grpc_fake_transport_security_credentials_create(); @@ -412,8 +410,8 @@ static void test_ssl_oauth2_iam_composite_creds(void) { grpc_credentials_unref(oauth2_creds); grpc_credentials_unref(aux_creds); grpc_credentials_unref(iam_creds); - GPR_ASSERT(strcmp(composite_creds->type, - GRPC_CREDENTIALS_TYPE_COMPOSITE) == 0); + GPR_ASSERT(strcmp(composite_creds->type, GRPC_CREDENTIALS_TYPE_COMPOSITE) == + 0); GPR_ASSERT(grpc_credentials_has_request_metadata(composite_creds)); GPR_ASSERT(!grpc_credentials_has_request_metadata_only(composite_creds)); creds_array = grpc_composite_credentials_get_credentials(composite_creds); @@ -424,7 +422,7 @@ static void test_ssl_oauth2_iam_composite_creds(void) { GRPC_CREDENTIALS_TYPE_OAUTH2) == 0); GPR_ASSERT(strcmp(creds_array->creds_array[2]->type, GRPC_CREDENTIALS_TYPE_IAM) == 0); - grpc_credentials_get_request_metadata(composite_creds, test_service_url, + grpc_credentials_get_request_metadata(composite_creds, NULL, test_service_url, check_ssl_oauth2_iam_composite_metadata, composite_creds); } @@ -455,9 +453,10 @@ static void validate_compute_engine_http_request( const grpc_httpcli_request *request) { GPR_ASSERT(!request->use_ssl); GPR_ASSERT(strcmp(request->host, "metadata") == 0); - GPR_ASSERT(strcmp(request->path, - "/computeMetadata/v1/instance/service-accounts/default/token") - == 0); + GPR_ASSERT( + strcmp(request->path, + "/computeMetadata/v1/instance/service-accounts/default/token") == + 0); GPR_ASSERT(request->hdr_count == 1); GPR_ASSERT(strcmp(request->hdrs[0].key, "Metadata-Flavor") == 0); GPR_ASSERT(strcmp(request->hdrs[0].value, "Google") == 0); @@ -506,16 +505,16 @@ static void test_compute_engine_creds_success(void) { /* First request: http get should be called. */ grpc_httpcli_set_override(compute_engine_httpcli_get_success_override, httpcli_post_should_not_be_called); - grpc_credentials_get_request_metadata(compute_engine_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + compute_engine_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); /* Second request: the cached token should be served directly. */ grpc_httpcli_set_override(httpcli_get_should_not_be_called, httpcli_post_should_not_be_called); - grpc_credentials_get_request_metadata(compute_engine_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + compute_engine_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); grpc_credentials_unref(compute_engine_creds); grpc_httpcli_set_override(NULL, NULL); @@ -528,9 +527,9 @@ static void test_compute_engine_creds_failure(void) { httpcli_post_should_not_be_called); GPR_ASSERT(grpc_credentials_has_request_metadata(compute_engine_creds)); GPR_ASSERT(grpc_credentials_has_request_metadata_only(compute_engine_creds)); - grpc_credentials_get_request_metadata(compute_engine_creds, test_service_url, - on_oauth2_creds_get_metadata_failure, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + compute_engine_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_failure, (void *)test_user_data); grpc_credentials_unref(compute_engine_creds); grpc_httpcli_set_override(NULL, NULL); } @@ -553,8 +552,8 @@ static void validate_refresh_token_http_request( GPR_ASSERT(strcmp(request->path, GRPC_GOOGLE_OAUTH2_SERVICE_TOKEN_PATH) == 0); GPR_ASSERT(request->hdr_count == 1); GPR_ASSERT(strcmp(request->hdrs[0].key, "Content-Type") == 0); - GPR_ASSERT(strcmp(request->hdrs[0].value, - "application/x-www-form-urlencoded") == 0); + GPR_ASSERT( + strcmp(request->hdrs[0].value, "application/x-www-form-urlencoded") == 0); } static int refresh_token_httpcli_post_success( @@ -587,16 +586,16 @@ static void test_refresh_token_creds_success(void) { /* First request: http get should be called. */ grpc_httpcli_set_override(httpcli_get_should_not_be_called, refresh_token_httpcli_post_success); - grpc_credentials_get_request_metadata(refresh_token_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + refresh_token_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); /* Second request: the cached token should be served directly. */ grpc_httpcli_set_override(httpcli_get_should_not_be_called, httpcli_post_should_not_be_called); - grpc_credentials_get_request_metadata(refresh_token_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + refresh_token_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); grpc_credentials_unref(refresh_token_creds); grpc_httpcli_set_override(NULL, NULL); @@ -609,9 +608,9 @@ static void test_refresh_token_creds_failure(void) { refresh_token_httpcli_post_failure); GPR_ASSERT(grpc_credentials_has_request_metadata(refresh_token_creds)); GPR_ASSERT(grpc_credentials_has_request_metadata_only(refresh_token_creds)); - grpc_credentials_get_request_metadata(refresh_token_creds, test_service_url, - on_oauth2_creds_get_metadata_failure, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + refresh_token_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_failure, (void *)test_user_data); grpc_credentials_unref(refresh_token_creds); grpc_httpcli_set_override(NULL, NULL); } @@ -667,8 +666,8 @@ static void validate_service_account_http_request( char *expected_body = NULL; GPR_ASSERT(body != NULL); GPR_ASSERT(body_size != 0); - gpr_asprintf(&expected_body, "%s%s", - GRPC_SERVICE_ACCOUNT_POST_BODY_PREFIX, test_signed_jwt); + gpr_asprintf(&expected_body, "%s%s", GRPC_SERVICE_ACCOUNT_POST_BODY_PREFIX, + test_signed_jwt); GPR_ASSERT(strlen(expected_body) == body_size); GPR_ASSERT(memcmp(expected_body, body, body_size) == 0); gpr_free(expected_body); @@ -677,8 +676,8 @@ static void validate_service_account_http_request( GPR_ASSERT(strcmp(request->path, GRPC_GOOGLE_OAUTH2_SERVICE_TOKEN_PATH) == 0); GPR_ASSERT(request->hdr_count == 1); GPR_ASSERT(strcmp(request->hdrs[0].key, "Content-Type") == 0); - GPR_ASSERT(strcmp(request->hdrs[0].value, - "application/x-www-form-urlencoded") == 0); + GPR_ASSERT( + strcmp(request->hdrs[0].value, "application/x-www-form-urlencoded") == 0); } static int service_account_httpcli_post_success( @@ -714,18 +713,18 @@ static void test_service_account_creds_success(void) { grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_success); grpc_httpcli_set_override(httpcli_get_should_not_be_called, service_account_httpcli_post_success); - grpc_credentials_get_request_metadata(service_account_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + service_account_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); /* Second request: the cached token should be served directly. */ grpc_jwt_encode_and_sign_set_override( encode_and_sign_jwt_should_not_be_called); grpc_httpcli_set_override(httpcli_get_should_not_be_called, httpcli_post_should_not_be_called); - grpc_credentials_get_request_metadata(service_account_creds, test_service_url, - on_oauth2_creds_get_metadata_success, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + service_account_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_success, (void *)test_user_data); gpr_free(json_key_string); grpc_credentials_unref(service_account_creds); @@ -744,9 +743,9 @@ static void test_service_account_creds_http_failure(void) { grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_success); grpc_httpcli_set_override(httpcli_get_should_not_be_called, service_account_httpcli_post_failure); - grpc_credentials_get_request_metadata(service_account_creds, test_service_url, - on_oauth2_creds_get_metadata_failure, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + service_account_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_failure, (void *)test_user_data); gpr_free(json_key_string); grpc_credentials_unref(service_account_creds); @@ -764,9 +763,9 @@ static void test_service_account_creds_signing_failure(void) { grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_failure); grpc_httpcli_set_override(httpcli_get_should_not_be_called, httpcli_post_should_not_be_called); - grpc_credentials_get_request_metadata(service_account_creds, test_service_url, - on_oauth2_creds_get_metadata_failure, - (void *)test_user_data); + grpc_credentials_get_request_metadata( + service_account_creds, NULL, test_service_url, + on_oauth2_creds_get_metadata_failure, (void *)test_user_data); gpr_free(json_key_string); grpc_credentials_unref(service_account_creds); @@ -808,21 +807,21 @@ static void test_jwt_creds_success(void) { /* First request: jwt_encode_and_sign should be called. */ grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_success); - grpc_credentials_get_request_metadata(jwt_creds, test_service_url, + grpc_credentials_get_request_metadata(jwt_creds, NULL, test_service_url, on_jwt_creds_get_metadata_success, (void *)test_user_data); /* Second request: the cached token should be served directly. */ grpc_jwt_encode_and_sign_set_override( encode_and_sign_jwt_should_not_be_called); - grpc_credentials_get_request_metadata(jwt_creds, test_service_url, + grpc_credentials_get_request_metadata(jwt_creds, NULL, test_service_url, on_jwt_creds_get_metadata_success, (void *)test_user_data); /* Third request: Different service url so jwt_encode_and_sign should be called again (no caching). */ grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_success); - grpc_credentials_get_request_metadata(jwt_creds, other_test_service_url, + grpc_credentials_get_request_metadata(jwt_creds, NULL, other_test_service_url, on_jwt_creds_get_metadata_success, (void *)test_user_data); @@ -839,7 +838,7 @@ static void test_jwt_creds_signing_failure(void) { GPR_ASSERT(grpc_credentials_has_request_metadata_only(jwt_creds)); grpc_jwt_encode_and_sign_set_override(encode_and_sign_jwt_failure); - grpc_credentials_get_request_metadata(jwt_creds, test_service_url, + grpc_credentials_get_request_metadata(jwt_creds, NULL, test_service_url, on_jwt_creds_get_metadata_failure, (void *)test_user_data); diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c index 30b23624d8..a8368fc842 100644 --- a/test/core/security/secure_endpoint_test.c +++ b/test/core/security/secure_endpoint_test.c @@ -44,6 +44,8 @@ #include "test/core/util/test_config.h" #include "src/core/tsi/fake_transport_security.h" +static grpc_pollset g_pollset; + static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair( size_t slice_size, gpr_slice *leftover_slices, size_t leftover_nslices) { tsi_frame_protector *fake_read_protector = tsi_create_fake_protector(NULL); @@ -52,6 +54,8 @@ static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair( grpc_endpoint_pair tcp; tcp = grpc_iomgr_create_endpoint_pair("fixture", slice_size); + grpc_endpoint_add_to_pollset(tcp.client, &g_pollset); + grpc_endpoint_add_to_pollset(tcp.server, &g_pollset); if (leftover_nslices == 0) { f.client_ep = @@ -190,13 +194,17 @@ static void test_destroy_ep_early(grpc_endpoint_test_config config, clean_up(); } +static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } + int main(int argc, char **argv) { grpc_test_init(argc, argv); grpc_iomgr_init(); - grpc_endpoint_tests(configs[0]); + grpc_pollset_init(&g_pollset); + grpc_endpoint_tests(configs[0], &g_pollset); test_leftover(configs[1], 1); test_destroy_ep_early(configs[1], 1); + grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset); grpc_iomgr_shutdown(); return 0; diff --git a/test/core/surface/byte_buffer_reader_test.c b/test/core/surface/byte_buffer_reader_test.c index 87a2cd7a43..7c2cb9484a 100644 --- a/test/core/surface/byte_buffer_reader_test.c +++ b/test/core/surface/byte_buffer_reader_test.c @@ -127,11 +127,11 @@ static void read_compressed_slice(grpc_compression_algorithm algorithm, input_slice = gpr_slice_malloc(input_size); memset(GPR_SLICE_START_PTR(input_slice), 'a', input_size); - gpr_slice_buffer_add(&sliceb_in, input_slice); /* takes ownership */ + gpr_slice_buffer_add(&sliceb_in, input_slice); /* takes ownership */ GPR_ASSERT(grpc_msg_compress(algorithm, &sliceb_in, &sliceb_out)); - buffer = grpc_raw_compressed_byte_buffer_create( - sliceb_out.slices, sliceb_out.count, algorithm); + buffer = grpc_raw_compressed_byte_buffer_create(sliceb_out.slices, + sliceb_out.count, algorithm); grpc_byte_buffer_reader_init(&reader, buffer); while (grpc_byte_buffer_reader_next(&reader, &read_slice)) { diff --git a/test/core/surface/completion_queue_test.c b/test/core/surface/completion_queue_test.c index 9e7b2ea1df..eba24f5c6e 100644 --- a/test/core/surface/completion_queue_test.c +++ b/test/core/surface/completion_queue_test.c @@ -100,7 +100,8 @@ static void test_shutdown_then_next_polling(void) { cc = grpc_completion_queue_create(); grpc_completion_queue_shutdown(cc); - GPR_ASSERT(grpc_completion_queue_next(cc, gpr_inf_past).type == GRPC_QUEUE_SHUTDOWN); + GPR_ASSERT(grpc_completion_queue_next(cc, gpr_inf_past).type == + GRPC_QUEUE_SHUTDOWN); grpc_completion_queue_destroy(cc); } @@ -110,7 +111,8 @@ static void test_shutdown_then_next_with_timeout(void) { cc = grpc_completion_queue_create(); grpc_completion_queue_shutdown(cc); - GPR_ASSERT(grpc_completion_queue_next(cc, gpr_inf_future).type == GRPC_QUEUE_SHUTDOWN); + GPR_ASSERT(grpc_completion_queue_next(cc, gpr_inf_future).type == + GRPC_QUEUE_SHUTDOWN); grpc_completion_queue_destroy(cc); } @@ -177,7 +179,7 @@ static void producer_thread(void *arg) { int i; gpr_log(GPR_INFO, "producer %d started", opt->id); - gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_started, (void *)(gpr_intptr)1); GPR_ASSERT(gpr_event_wait(opt->phase1, ten_seconds_time())); gpr_log(GPR_INFO, "producer %d phase 1", opt->id); @@ -186,7 +188,7 @@ static void producer_thread(void *arg) { } gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id); - gpr_event_set(&opt->on_phase1_done, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_phase1_done, (void *)(gpr_intptr)1); GPR_ASSERT(gpr_event_wait(opt->phase2, ten_seconds_time())); gpr_log(GPR_INFO, "producer %d phase 2", opt->id); @@ -196,7 +198,7 @@ static void producer_thread(void *arg) { } gpr_log(GPR_INFO, "producer %d phase 2 done", opt->id); - gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_finished, (void *)(gpr_intptr)1); } static void consumer_thread(void *arg) { @@ -204,13 +206,13 @@ static void consumer_thread(void *arg) { grpc_event ev; gpr_log(GPR_INFO, "consumer %d started", opt->id); - gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_started, (void *)(gpr_intptr)1); GPR_ASSERT(gpr_event_wait(opt->phase1, ten_seconds_time())); gpr_log(GPR_INFO, "consumer %d phase 1", opt->id); gpr_log(GPR_INFO, "consumer %d phase 1 done", opt->id); - gpr_event_set(&opt->on_phase1_done, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_phase1_done, (void *)(gpr_intptr)1); GPR_ASSERT(gpr_event_wait(opt->phase2, ten_seconds_time())); gpr_log(GPR_INFO, "consumer %d phase 2", opt->id); @@ -223,7 +225,7 @@ static void consumer_thread(void *arg) { break; case GRPC_QUEUE_SHUTDOWN: gpr_log(GPR_INFO, "consumer %d phase 2 done", opt->id); - gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1); + gpr_event_set(&opt->on_finished, (void *)(gpr_intptr)1); return; case GRPC_QUEUE_TIMEOUT: gpr_log(GPR_ERROR, "Invalid timeout received"); @@ -242,10 +244,8 @@ static void test_threading(int producers, int consumers) { int total_consumed = 0; static int optid = 101; - gpr_log(GPR_INFO, "%s: %d producers, %d consumers", "test_threading", producers, - consumers); - - grpc_completion_queue_dont_poll_test_only(cc); + gpr_log(GPR_INFO, "%s: %d producers, %d consumers", "test_threading", + producers, consumers); /* start all threads: they will wait for phase1 */ for (i = 0; i < producers + consumers; i++) { @@ -267,7 +267,7 @@ static void test_threading(int producers, int consumers) { /* start phase1: producers will pre-declare all operations they will complete */ gpr_log(GPR_INFO, "start phase 1"); - gpr_event_set(&phase1, (void *)(gpr_intptr) 1); + gpr_event_set(&phase1, (void *)(gpr_intptr)1); gpr_log(GPR_INFO, "wait phase 1"); for (i = 0; i < producers + consumers; i++) { @@ -277,7 +277,7 @@ static void test_threading(int producers, int consumers) { /* start phase2: operations will complete, and consumers will consume them */ gpr_log(GPR_INFO, "start phase 2"); - gpr_event_set(&phase2, (void *)(gpr_intptr) 1); + gpr_event_set(&phase2, (void *)(gpr_intptr)1); /* in parallel, we shutdown the completion channel - all events should still be consumed */ diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 871f956491..117d8bb9fa 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -67,27 +67,45 @@ namespace { void* tag(int i) { return (void*)(gpr_intptr) i; } -void verify_ok(CompletionQueue* cq, int i, bool expect_ok) { - bool ok; - void* got_tag; - EXPECT_TRUE(cq->Next(&got_tag, &ok)); - EXPECT_EQ(expect_ok, ok); - EXPECT_EQ(tag(i), got_tag); -} - -void verify_timed_ok( - CompletionQueue* cq, int i, bool expect_ok, - std::chrono::system_clock::time_point deadline = - std::chrono::system_clock::time_point::max(), - CompletionQueue::NextStatus expected_outcome = CompletionQueue::GOT_EVENT) { - bool ok; - void* got_tag; - EXPECT_EQ(cq->AsyncNext(&got_tag, &ok, deadline), expected_outcome); - if (expected_outcome == CompletionQueue::GOT_EVENT) { - EXPECT_EQ(expect_ok, ok); - EXPECT_EQ(tag(i), got_tag); +class Verifier { + public: + Verifier& Expect(int i, bool expect_ok) { + expectations_[tag(i)] = expect_ok; + return *this; } -} + void Verify(CompletionQueue *cq) { + GPR_ASSERT(!expectations_.empty()); + while (!expectations_.empty()) { + bool ok; + void* got_tag; + EXPECT_TRUE(cq->Next(&got_tag, &ok)); + auto it = expectations_.find(got_tag); + EXPECT_TRUE(it != expectations_.end()); + EXPECT_EQ(it->second, ok); + expectations_.erase(it); + } + } + void Verify(CompletionQueue *cq, std::chrono::system_clock::time_point deadline) { + if (expectations_.empty()) { + bool ok; + void *got_tag; + EXPECT_EQ(cq->AsyncNext(&got_tag, &ok, deadline), CompletionQueue::TIMEOUT); + } else { + while (!expectations_.empty()) { + bool ok; + void *got_tag; + EXPECT_EQ(cq->AsyncNext(&got_tag, &ok, deadline), CompletionQueue::GOT_EVENT); + auto it = expectations_.find(got_tag); + EXPECT_TRUE(it != expectations_.end()); + EXPECT_EQ(it->second, ok); + expectations_.erase(it); + } + } + } + + private: + std::map<void*, bool> expectations_; +}; class AsyncEnd2endTest : public ::testing::Test { protected: @@ -100,7 +118,7 @@ class AsyncEnd2endTest : public ::testing::Test { ServerBuilder builder; builder.AddListeningPort(server_address_.str(), grpc::InsecureServerCredentials()); builder.RegisterAsyncService(&service_); - srv_cq_ = builder.AddCompletionQueue(); + cq_ = builder.AddCompletionQueue(); server_ = builder.BuildAndStart(); } @@ -108,11 +126,8 @@ class AsyncEnd2endTest : public ::testing::Test { server_->Shutdown(); void* ignored_tag; bool ignored_ok; - cli_cq_.Shutdown(); - srv_cq_->Shutdown(); - while (cli_cq_.Next(&ignored_tag, &ignored_ok)) - ; - while (srv_cq_->Next(&ignored_tag, &ignored_ok)) + cq_->Shutdown(); + while (cq_->Next(&ignored_tag, &ignored_ok)) ; } @@ -122,11 +137,6 @@ class AsyncEnd2endTest : public ::testing::Test { stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel)); } - void server_ok(int i) { verify_ok(srv_cq_.get(), i, true); } - void client_ok(int i) { verify_ok(&cli_cq_, i, true); } - void server_fail(int i) { verify_ok(srv_cq_.get(), i, false); } - void client_fail(int i) { verify_ok(&cli_cq_, i, false); } - void SendRpc(int num_rpcs) { for (int i = 0; i < num_rpcs; i++) { EchoRequest send_request; @@ -141,28 +151,27 @@ class AsyncEnd2endTest : public ::testing::Test { send_request.set_message("Hello"); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, - srv_cq_.get(), srv_cq_.get(), tag(2)); + cq_.get(), cq_.get(), tag(2)); - server_ok(2); + Verifier().Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); send_response.set_message(recv_request.message()); response_writer.Finish(send_response, Status::OK, tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); response_reader->Finish(&recv_response, &recv_status, tag(4)); - client_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); } } - CompletionQueue cli_cq_; - std::unique_ptr<ServerCompletionQueue> srv_cq_; + std::unique_ptr<ServerCompletionQueue> cq_; std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_; std::unique_ptr<Server> server_; grpc::cpp::test::util::TestService::AsyncService service_; @@ -195,27 +204,27 @@ TEST_F(AsyncEnd2endTest, AsyncNextRpc) { send_request.set_message("Hello"); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); std::chrono::system_clock::time_point time_now( std::chrono::system_clock::now()); std::chrono::system_clock::time_point time_limit( std::chrono::system_clock::now() + std::chrono::seconds(10)); - verify_timed_ok(srv_cq_.get(), -1, true, time_now, CompletionQueue::TIMEOUT); - verify_timed_ok(&cli_cq_, -1, true, time_now, CompletionQueue::TIMEOUT); + Verifier().Verify(cq_.get(), time_now); + Verifier().Verify(cq_.get(), time_now); - service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(), - srv_cq_.get(), tag(2)); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); - verify_timed_ok(srv_cq_.get(), 2, true, time_limit); + Verifier().Expect(2, true).Verify(cq_.get(), time_limit); EXPECT_EQ(send_request.message(), recv_request.message()); send_response.set_message(recv_request.message()); response_writer.Finish(send_response, Status::OK, tag(3)); - verify_timed_ok(srv_cq_.get(), 3, true); + Verifier().Expect(3, true).Verify(cq_.get(), std::chrono::system_clock::time_point::max()); response_reader->Finish(&recv_response, &recv_status, tag(4)); - verify_timed_ok(&cli_cq_, 4, true); + Verifier().Expect(4, true).Verify(cq_.get(), std::chrono::system_clock::time_point::max()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); @@ -236,40 +245,39 @@ TEST_F(AsyncEnd2endTest, SimpleClientStreaming) { send_request.set_message("Hello"); std::unique_ptr<ClientAsyncWriter<EchoRequest> > cli_stream( - stub_->AsyncRequestStream(&cli_ctx, &recv_response, &cli_cq_, tag(1))); + stub_->AsyncRequestStream(&cli_ctx, &recv_response, cq_.get(), tag(1))); - service_.RequestRequestStream(&srv_ctx, &srv_stream, srv_cq_.get(), - srv_cq_.get(), tag(2)); + service_.RequestRequestStream(&srv_ctx, &srv_stream, cq_.get(), + cq_.get(), tag(2)); - server_ok(2); - client_ok(1); + Verifier().Expect(2, true).Expect(1, true).Verify(cq_.get()); cli_stream->Write(send_request, tag(3)); - client_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); srv_stream.Read(&recv_request, tag(4)); - server_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); cli_stream->Write(send_request, tag(5)); - client_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); srv_stream.Read(&recv_request, tag(6)); - server_ok(6); + Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); cli_stream->WritesDone(tag(7)); - client_ok(7); + Verifier().Expect(7, true).Verify(cq_.get()); srv_stream.Read(&recv_request, tag(8)); - server_fail(8); + Verifier().Expect(8, false).Verify(cq_.get()); send_response.set_message(recv_request.message()); srv_stream.Finish(send_response, Status::OK, tag(9)); - server_ok(9); + Verifier().Expect(9, true).Verify(cq_.get()); cli_stream->Finish(&recv_status, tag(10)); - client_ok(10); + Verifier().Expect(10, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); @@ -290,38 +298,37 @@ TEST_F(AsyncEnd2endTest, SimpleServerStreaming) { send_request.set_message("Hello"); std::unique_ptr<ClientAsyncReader<EchoResponse> > cli_stream( - stub_->AsyncResponseStream(&cli_ctx, send_request, &cli_cq_, tag(1))); + stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1))); service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream, - srv_cq_.get(), srv_cq_.get(), tag(2)); + cq_.get(), cq_.get(), tag(2)); - server_ok(2); - client_ok(1); + Verifier().Expect(1, true).Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); send_response.set_message(recv_request.message()); srv_stream.Write(send_response, tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); cli_stream->Read(&recv_response, tag(4)); - client_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); srv_stream.Write(send_response, tag(5)); - server_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); cli_stream->Read(&recv_response, tag(6)); - client_ok(6); + Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); srv_stream.Finish(Status::OK, tag(7)); - server_ok(7); + Verifier().Expect(7, true).Verify(cq_.get()); cli_stream->Read(&recv_response, tag(8)); - client_fail(8); + Verifier().Expect(8, false).Verify(cq_.get()); cli_stream->Finish(&recv_status, tag(9)); - client_ok(9); + Verifier().Expect(9, true).Verify(cq_.get()); EXPECT_TRUE(recv_status.ok()); } @@ -341,40 +348,39 @@ TEST_F(AsyncEnd2endTest, SimpleBidiStreaming) { send_request.set_message("Hello"); std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse> > - cli_stream(stub_->AsyncBidiStream(&cli_ctx, &cli_cq_, tag(1))); + cli_stream(stub_->AsyncBidiStream(&cli_ctx, cq_.get(), tag(1))); - service_.RequestBidiStream(&srv_ctx, &srv_stream, srv_cq_.get(), - srv_cq_.get(), tag(2)); + service_.RequestBidiStream(&srv_ctx, &srv_stream, cq_.get(), + cq_.get(), tag(2)); - server_ok(2); - client_ok(1); + Verifier().Expect(1, true).Expect(2, true).Verify(cq_.get()); cli_stream->Write(send_request, tag(3)); - client_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); srv_stream.Read(&recv_request, tag(4)); - server_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); send_response.set_message(recv_request.message()); srv_stream.Write(send_response, tag(5)); - server_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); cli_stream->Read(&recv_response, tag(6)); - client_ok(6); + Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); cli_stream->WritesDone(tag(7)); - client_ok(7); + Verifier().Expect(7, true).Verify(cq_.get()); srv_stream.Read(&recv_request, tag(8)); - server_fail(8); + Verifier().Expect(8, false).Verify(cq_.get()); srv_stream.Finish(Status::OK, tag(9)); - server_ok(9); + Verifier().Expect(9, true).Verify(cq_.get()); cli_stream->Finish(&recv_status, tag(10)); - client_ok(10); + Verifier().Expect(10, true).Verify(cq_.get()); EXPECT_TRUE(recv_status.ok()); } @@ -400,11 +406,11 @@ TEST_F(AsyncEnd2endTest, ClientInitialMetadataRpc) { cli_ctx.AddMetadata(meta2.first, meta2.second); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); - service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(), - srv_cq_.get(), tag(2)); - server_ok(2); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + Verifier().Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); auto client_initial_metadata = srv_ctx.client_metadata(); EXPECT_EQ(meta1.second, client_initial_metadata.find(meta1.first)->second); @@ -414,10 +420,10 @@ TEST_F(AsyncEnd2endTest, ClientInitialMetadataRpc) { send_response.set_message(recv_request.message()); response_writer.Finish(send_response, Status::OK, tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); response_reader->Finish(&recv_response, &recv_status, tag(4)); - client_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); @@ -441,19 +447,19 @@ TEST_F(AsyncEnd2endTest, ServerInitialMetadataRpc) { std::pair<grpc::string, grpc::string> meta2("key2", "val2"); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); - service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(), - srv_cq_.get(), tag(2)); - server_ok(2); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + Verifier().Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); srv_ctx.AddInitialMetadata(meta1.first, meta1.second); srv_ctx.AddInitialMetadata(meta2.first, meta2.second); response_writer.SendInitialMetadata(tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); response_reader->ReadInitialMetadata(tag(4)); - client_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); auto server_initial_metadata = cli_ctx.GetServerInitialMetadata(); EXPECT_EQ(meta1.second, server_initial_metadata.find(meta1.first)->second); EXPECT_EQ(meta2.second, server_initial_metadata.find(meta2.first)->second); @@ -461,10 +467,10 @@ TEST_F(AsyncEnd2endTest, ServerInitialMetadataRpc) { send_response.set_message(recv_request.message()); response_writer.Finish(send_response, Status::OK, tag(5)); - server_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); response_reader->Finish(&recv_response, &recv_status, tag(6)); - client_ok(6); + Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); @@ -488,24 +494,24 @@ TEST_F(AsyncEnd2endTest, ServerTrailingMetadataRpc) { std::pair<grpc::string, grpc::string> meta2("key2", "val2"); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); - service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(), - srv_cq_.get(), tag(2)); - server_ok(2); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + Verifier().Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); response_writer.SendInitialMetadata(tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); send_response.set_message(recv_request.message()); srv_ctx.AddTrailingMetadata(meta1.first, meta1.second); srv_ctx.AddTrailingMetadata(meta2.first, meta2.second); response_writer.Finish(send_response, Status::OK, tag(4)); - server_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); response_reader->Finish(&recv_response, &recv_status, tag(5)); - client_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata(); @@ -548,11 +554,11 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { cli_ctx.AddMetadata(meta2.first, meta2.second); std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader( - stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_)); + stub_->AsyncEcho(&cli_ctx, send_request, cq_.get())); - service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(), - srv_cq_.get(), tag(2)); - server_ok(2); + service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq_.get(), + cq_.get(), tag(2)); + Verifier().Expect(2, true).Verify(cq_.get()); EXPECT_EQ(send_request.message(), recv_request.message()); auto client_initial_metadata = srv_ctx.client_metadata(); EXPECT_EQ(meta1.second, client_initial_metadata.find(meta1.first)->second); @@ -562,9 +568,9 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { srv_ctx.AddInitialMetadata(meta3.first, meta3.second); srv_ctx.AddInitialMetadata(meta4.first, meta4.second); response_writer.SendInitialMetadata(tag(3)); - server_ok(3); + Verifier().Expect(3, true).Verify(cq_.get()); response_reader->ReadInitialMetadata(tag(4)); - client_ok(4); + Verifier().Expect(4, true).Verify(cq_.get()); auto server_initial_metadata = cli_ctx.GetServerInitialMetadata(); EXPECT_EQ(meta3.second, server_initial_metadata.find(meta3.first)->second); EXPECT_EQ(meta4.second, server_initial_metadata.find(meta4.first)->second); @@ -575,10 +581,10 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { srv_ctx.AddTrailingMetadata(meta6.first, meta6.second); response_writer.Finish(send_response, Status::OK, tag(5)); - server_ok(5); + Verifier().Expect(5, true).Verify(cq_.get()); response_reader->Finish(&recv_response, &recv_status, tag(6)); - client_ok(6); + Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); EXPECT_TRUE(recv_status.ok()); auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata(); diff --git a/test/cpp/end2end/client_crash_test.cc b/test/cpp/end2end/client_crash_test.cc index aeba295947..7876e8dee3 100644 --- a/test/cpp/end2end/client_crash_test.cc +++ b/test/cpp/end2end/client_crash_test.cc @@ -90,15 +90,13 @@ class CrashTest : public ::testing::Test { void KillServer() { server_.reset(); - // give some time for the TCP connection to drop - gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(1))); } private: std::unique_ptr<SubProcess> server_; }; -TEST_F(CrashTest, KillAfterWrite) { +TEST_F(CrashTest, KillBeforeWrite) { auto stub = CreateServerAndStub(); EchoRequest request; @@ -112,17 +110,18 @@ TEST_F(CrashTest, KillAfterWrite) { EXPECT_TRUE(stream->Read(&response)); EXPECT_EQ(response.message(), request.message()); - request.set_message("I'm going to kill you"); - EXPECT_TRUE(stream->Write(request)); - KillServer(); + request.set_message("You should be dead"); + // This may succeed or fail depending on the state of the TCP connection + stream->Write(request); + // But the read will definitely fail EXPECT_FALSE(stream->Read(&response)); EXPECT_FALSE(stream->Finish().ok()); } -TEST_F(CrashTest, KillBeforeWrite) { +TEST_F(CrashTest, KillAfterWrite) { auto stub = CreateServerAndStub(); EchoRequest request; @@ -136,10 +135,11 @@ TEST_F(CrashTest, KillBeforeWrite) { EXPECT_TRUE(stream->Read(&response)); EXPECT_EQ(response.message(), request.message()); + request.set_message("I'm going to kill you"); + EXPECT_TRUE(stream->Write(request)); + KillServer(); - request.set_message("You should be dead"); - EXPECT_FALSE(stream->Write(request)); EXPECT_FALSE(stream->Read(&response)); EXPECT_FALSE(stream->Finish().ok()); @@ -161,5 +161,11 @@ int main(int argc, char** argv) { grpc_test_init(argc, argv); ::testing::InitGoogleTest(&argc, argv); - return RUN_ALL_TESTS(); + // Order seems to matter on these tests: run three times to eliminate that + for (int i = 0; i < 3; i++) { + if (RUN_ALL_TESTS() != 0) { + return 1; + } + } + return 0; } diff --git a/test/cpp/qps/qps_test_with_poll.cc b/test/cpp/qps/qps_test_with_poll.cc new file mode 100644 index 0000000000..90a8da8d11 --- /dev/null +++ b/test/cpp/qps/qps_test_with_poll.cc @@ -0,0 +1,90 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <set> + +#include <grpc/support/log.h> + +#include <signal.h> + +#include "test/cpp/qps/driver.h" +#include "test/cpp/qps/report.h" +#include "test/cpp/util/benchmark_config.h" + +extern "C" { +#include "src/core/iomgr/pollset_posix.h" +} + +namespace grpc { +namespace testing { + +static const int WARMUP = 5; +static const int BENCHMARK = 5; + +static void RunQPS() { + gpr_log(GPR_INFO, "Running QPS test"); + + ClientConfig client_config; + client_config.set_client_type(ASYNC_CLIENT); + client_config.set_enable_ssl(false); + client_config.set_outstanding_rpcs_per_channel(1000); + client_config.set_client_channels(8); + client_config.set_payload_size(1); + client_config.set_async_client_threads(8); + client_config.set_rpc_type(UNARY); + + ServerConfig server_config; + server_config.set_server_type(ASYNC_SERVER); + server_config.set_enable_ssl(false); + server_config.set_threads(4); + + const auto result = + RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); + + GetReporter()->ReportQPSPerCore(*result); + GetReporter()->ReportLatency(*result); +} + +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc::testing::InitBenchmark(&argc, &argv, true); + + grpc_platform_become_multipoller = grpc_poll_become_multipoller; + + signal(SIGPIPE, SIG_IGN); + grpc::testing::RunQPS(); + + return 0; +} |