diff options
Diffstat (limited to 'test')
58 files changed, 997 insertions, 1073 deletions
diff --git a/test/core/bad_client/bad_client.c b/test/core/bad_client/bad_client.c index 7319545c84..e9adcf34c7 100644 --- a/test/core/bad_client/bad_client.c +++ b/test/core/bad_client/bad_client.c @@ -143,6 +143,10 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator, if (sfd.client) { grpc_endpoint_destroy(sfd.client); } + grpc_server_shutdown_and_notify(a.server, a.cq, NULL); + GPR_ASSERT(grpc_completion_queue_pluck(a.cq, NULL, + GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1)) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(a.server); grpc_completion_queue_destroy(a.cq); 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 06614a93e7..1142c37ea4 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); @@ -153,10 +149,10 @@ void test_connect(const char *server_host, const char *client_host, int port, /* 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); + &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; @@ -173,11 +169,9 @@ 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")); @@ -188,29 +182,26 @@ void test_connect(const char *server_host, const char *client_host, int port, 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(server); + grpc_server_shutdown_and_notify(server, cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).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..8996bacd24 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; diff --git a/test/core/end2end/fixtures/chttp2_fake_security.c b/test/core/end2end/fixtures/chttp2_fake_security.c index 5323e29e82..7e98d729b3 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,7 +82,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); 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..e351fb3241 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,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); 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..77c057df2c 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,9 +85,8 @@ 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); } @@ -144,7 +142,7 @@ int main(int argc, char **argv) { grpc_test_init(argc, argv); /* Set the SSL roots env var. */ - roots_file = gpr_tmpfile("chttp2_simple_ssl_with_poll_fullstack_test", + roots_file = gpr_tmpfile("chttp2_simple_ssl_with_poll_fullstack_test", &roots_filename); GPR_ASSERT(roots_filename != NULL); GPR_ASSERT(roots_file != NULL); 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..8d512635f9 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,7 +83,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); 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 fb13c7840c..7d581f64cc 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 0220f34534..09b20f4a89 100644 --- a/test/core/end2end/tests/bad_hostname.c +++ b/test/core/end2end/tests/bad_hostname.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,18 +92,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; @@ -113,7 +111,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "slartibartfast.local", deadline); GPR_ASSERT(c); @@ -139,8 +137,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); @@ -152,7 +150,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 74bbd014c7..fedcea6685 100644 --- a/test/core/end2end/tests/cancel_after_accept.c +++ b/test/core/end2end/tests/cancel_after_accept.c @@ -75,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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 +91,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 */ @@ -107,8 +105,7 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, grpc_call *s; 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; @@ -126,7 +123,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -158,10 +155,10 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, 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); + &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; @@ -180,11 +177,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)); @@ -204,8 +199,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 945cba7ee0..c31bd7fa26 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,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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 +91,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 */ @@ -107,8 +105,7 @@ static void test_cancel_after_accept_and_writes_closed( grpc_call *s; 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; @@ -126,7 +123,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -160,10 +157,10 @@ static void test_cancel_after_accept_and_writes_closed( 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); + &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; @@ -182,11 +179,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)); @@ -206,8 +201,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 c019709bd7..2b66e5a37e 100644 --- a/test/core/end2end/tests/cancel_after_invoke.c +++ b/test/core/end2end/tests/cancel_after_invoke.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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 +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 after invoke, no payload */ @@ -108,7 +106,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; @@ -121,7 +119,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -155,8 +153,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)); @@ -172,7 +170,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 3d901f18a2..697f347066 100644 --- a/test/core/end2end/tests/cancel_before_invoke.c +++ b/test/core/end2end/tests/cancel_before_invoke.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Cancel before invoke */ @@ -105,7 +103,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config, grpc_call *c; 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; @@ -118,7 +116,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -152,8 +150,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); @@ -168,7 +166,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 f0984cb5dc..7c864016b8 100644 --- a/test/core/end2end/tests/cancel_in_a_vacuum.c +++ b/test/core/end2end/tests/cancel_in_a_vacuum.c @@ -46,6 +46,8 @@ enum { TIMEOUT = 200000 }; +static void *tag(gpr_intptr t) { return (void *)t; } + static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, const char *test_name, grpc_channel_args *client_args, @@ -73,7 +75,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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 +91,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 and do nothing */ @@ -102,9 +102,9 @@ static void test_cancel_in_a_vacuum(grpc_end2end_test_config config, grpc_call *c; 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c index e0f996993f..f3a46e23c9 100644 --- a/test/core/end2end/tests/census_simple_request.c +++ b/test/core/end2end/tests/census_simple_request.c @@ -61,9 +61,12 @@ static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, return f; } +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(f->server); + 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; } @@ -85,22 +88,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 *tag(gpr_intptr t) { return (void *)t; } - 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; @@ -112,7 +109,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); @@ -140,10 +137,10 @@ static void test_body(grpc_end2end_test_fixture f) { 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); + &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; @@ -159,11 +156,9 @@ 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")); @@ -180,8 +175,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 60e7d227b9..d962f870b0 100644 --- a/test/core/end2end/tests/disappearing_server.c +++ b/test/core/end2end/tests/disappearing_server.c @@ -62,7 +62,6 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); grpc_server_destroy(f->server); f->server = NULL; } @@ -77,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(); @@ -102,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); @@ -130,14 +125,14 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, 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); + &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(f->server); + grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -153,11 +148,10 @@ 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_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")); @@ -177,23 +171,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 a44823033d..ca07024560 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 @@ -72,13 +72,6 @@ static void drain_cq(grpc_completion_queue *cq) { } while (ev.type != GRPC_QUEUE_SHUTDOWN); } -static void shutdown_server(grpc_end2end_test_fixture *f) { - if (!f->server) return; - grpc_server_shutdown(f->server); - grpc_server_destroy(f->server); - f->server = NULL; -} - static void shutdown_client(grpc_end2end_test_fixture *f) { if (!f->client) return; grpc_channel_destroy(f->client); @@ -86,15 +79,11 @@ static void shutdown_client(grpc_end2end_test_fixture *f) { } 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( @@ -103,8 +92,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -116,7 +104,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -145,10 +133,10 @@ static void test_early_server_shutdown_finishes_inflight_calls( 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); + &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; @@ -157,13 +145,15 @@ 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 */ - shutdown_server(&f); + 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_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); - cq_expect_completion(v_client, tag(1), 1); - cq_verify(v_client); + grpc_server_destroy(f.server); GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); @@ -179,8 +169,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 a8eb2144bb..c1f475d6fb 100644 --- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c +++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c @@ -72,13 +72,6 @@ static void drain_cq(grpc_completion_queue *cq) { } while (ev.type != GRPC_QUEUE_SHUTDOWN); } -static void shutdown_server(grpc_end2end_test_fixture *f) { - if (!f->server) return; - /* don't shutdown, just destroy, to tickle this code edge */ - grpc_server_destroy(f->server); - f->server = NULL; -} - static void shutdown_client(grpc_end2end_test_fixture *f) { if (!f->client) return; grpc_channel_destroy(f->client); @@ -86,21 +79,17 @@ static void shutdown_client(grpc_end2end_test_fixture *f) { } 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_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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_call *s = (void *)1; grpc_call_details call_details; grpc_metadata_array request_metadata_recv; @@ -112,16 +101,19 @@ static void test_early_server_shutdown_finishes_tags( 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(f.server); - cq_expect_completion(v_server, tag(101), 0); - cq_verify(v_server); + &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 d1e5527e9e..27366b715a 100644 --- a/test/core/end2end/tests/empty_batch.c +++ b/test/core/end2end/tests/empty_batch.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,31 +92,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", + 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 d7b9fde3a6..c1fc6fb4c5 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( @@ -102,8 +99,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -115,7 +111,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -144,14 +140,14 @@ static void test_early_server_shutdown_finishes_inflight_calls( 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); + &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, 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; @@ -167,15 +163,12 @@ 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_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_server, tag(0xdead), 1); - cq_verify(v_server); - - 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")); @@ -190,8 +183,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 a1750ed744..4c55198762 100644 --- a/test/core/end2end/tests/invoke_large_request.c +++ b/test/core/end2end/tests/invoke_large_request.c @@ -72,7 +72,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -87,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 gpr_slice large_slice(void) { @@ -113,8 +111,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; @@ -128,7 +125,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -162,10 +159,10 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { 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); + &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; @@ -176,8 +173,8 @@ static void test_invoke_large_request(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; @@ -186,7 +183,6 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { op->op = GRPC_OP_SEND_MESSAGE; op->data.send_message = response_payload; 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; @@ -194,11 +190,9 @@ static void test_invoke_large_request(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_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(details, "xyz")); @@ -215,8 +209,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 ef0af34c0d..8cfa15a9bc 100644 --- a/test/core/end2end/tests/max_concurrent_streams.c +++ b/test/core/end2end/tests/max_concurrent_streams.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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 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; @@ -114,7 +111,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); @@ -142,10 +139,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) { 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); + &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; @@ -161,11 +158,9 @@ 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")); @@ -182,8 +177,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) { @@ -196,8 +190,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; @@ -214,6 +207,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; @@ -223,8 +218,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); @@ -241,18 +235,18 @@ 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))); + &request_metadata_recv, f.cq, + f.cq, tag(101))); op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -298,18 +292,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; @@ -326,21 +330,19 @@ 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); + &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; @@ -357,14 +359,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 6f1a5815e9..66c2e37ec5 100644 --- a/test/core/end2end/tests/max_message_length.c +++ b/test/core/end2end/tests/max_message_length.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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 void test_max_message_length(grpc_end2end_test_config config) { @@ -103,8 +101,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"); @@ -127,10 +124,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); @@ -161,10 +157,10 @@ static void test_max_message_length(grpc_end2end_test_config config) { 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); + &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; @@ -172,11 +168,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")); @@ -193,8 +187,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 5b18efcbfd..e412870ad2 100644 --- a/test/core/end2end/tests/no_op.c +++ b/test/core/end2end/tests/no_op.c @@ -45,6 +45,8 @@ enum { TIMEOUT = 200000 }; +static void *tag(gpr_intptr t) { return (void *)t; } + static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, const char *test_name, grpc_channel_args *client_args, @@ -72,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -87,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 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 97ac9bb491..95b289b104 100644 --- a/test/core/end2end/tests/ping_pong_streaming.c +++ b/test/core/end2end/tests/ping_pong_streaming.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Client pings and server pongs. Repeat messages rounds before finishing. */ @@ -104,8 +102,7 @@ static void test_pingpong_streaming(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; @@ -124,7 +121,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); @@ -150,10 +147,10 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, 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); + &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; @@ -183,8 +180,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; @@ -192,11 +189,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); @@ -220,19 +215,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 2cf2ccec1a..70c70e9c40 100644 --- a/test/core/end2end/tests/registered_call.c +++ b/test/core/end2end/tests/registered_call.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,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, 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; @@ -116,7 +113,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); @@ -143,10 +140,10 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) { 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); + &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; @@ -162,11 +159,9 @@ 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")); @@ -183,8 +178,7 @@ 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) { 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 9788b70dfb..0d29b30a81 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,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Request/response with metadata and payload.*/ @@ -128,8 +126,7 @@ static void test_request_response_with_metadata_and_payload( 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -143,7 +140,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -178,10 +175,10 @@ static void test_request_response_with_metadata_and_payload( 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); + &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; @@ -193,8 +190,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; @@ -210,11 +207,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")); @@ -245,8 +240,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 fe61aef28c..f36e2e76fb 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,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Request/response with metadata and payload.*/ @@ -114,8 +112,7 @@ static void test_request_response_with_metadata_and_payload( 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -129,7 +126,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -164,10 +161,10 @@ static void test_request_response_with_metadata_and_payload( 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); + &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 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; @@ -196,11 +193,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")); @@ -223,8 +218,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 18054869e2..f44ee549a0 100644 --- a/test/core/end2end/tests/request_response_with_payload.c +++ b/test/core/end2end/tests/request_response_with_payload.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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 void request_response_with_payload(grpc_end2end_test_fixture f) { @@ -107,8 +105,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; @@ -122,7 +119,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -156,10 +153,10 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { 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); + &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,8 +167,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; @@ -187,11 +184,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")); @@ -210,8 +205,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); 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 641a40907b..6fb9b2662d 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 @@ -88,7 +88,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -103,12 +104,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) { @@ -133,7 +131,7 @@ static void test_call_creds_failure(grpc_end2end_test_config config) { grpc_credentials *creds = 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -162,8 +160,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; @@ -179,7 +176,7 @@ 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", + 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); @@ -232,10 +229,10 @@ static void request_response_with_payload_and_call_creds( GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, &call_details, &request_metadata_recv, - f.server_cq, f.server_cq, + f.cq, f.cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + 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); @@ -252,8 +249,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; @@ -269,11 +266,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")); @@ -325,8 +320,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); 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 8ff7f99ff4..0f65c82241 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,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Request/response with metadata and payload.*/ @@ -113,8 +111,7 @@ static void test_request_response_with_metadata_and_payload( 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -128,7 +125,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -164,10 +161,10 @@ static void test_request_response_with_metadata_and_payload( GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, &call_details, &request_metadata_recv, - f.server_cq, f.server_cq, + f.cq, f.cq, tag(101))); - cq_expect_completion(v_server, tag(101), 1); - cq_verify(v_server); + 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 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; @@ -197,11 +194,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")); @@ -225,8 +220,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_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c index 31406f57de..3abacf8ef9 100644 --- a/test/core/end2end/tests/request_with_large_metadata.c +++ b/test/core/end2end/tests/request_with_large_metadata.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Request with a large amount of metadata.*/ @@ -107,8 +105,7 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -122,7 +119,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -160,10 +157,10 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { 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); + &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; @@ -174,8 +171,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; @@ -188,11 +185,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")); @@ -211,8 +206,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 ea4bb2cf91..8972d7fea1 100644 --- a/test/core/end2end/tests/request_with_payload.c +++ b/test/core/end2end/tests/request_with_payload.c @@ -74,7 +74,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -89,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); } /* Client sends a request with payload, server reads then returns status. */ @@ -106,8 +104,7 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { 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); + cq_verifier *cqv = cq_verifier_create(f.cq); grpc_op ops[6]; grpc_op *op; grpc_metadata_array initial_metadata_recv; @@ -120,7 +117,7 @@ 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", + c = grpc_channel_create_call(f.client, f.cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -151,10 +148,10 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { 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); + &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,8 +162,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; @@ -179,11 +176,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")); @@ -201,8 +196,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 a0c1865290..fe8657a2cc 100644 --- a/test/core/end2end/tests/server_finishes_request.c +++ b/test/core/end2end/tests/server_finishes_request.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,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; @@ -116,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); @@ -142,10 +139,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) { 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); + &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; @@ -161,11 +158,9 @@ 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")); @@ -182,8 +177,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 59cc9b5488..f399d0ea9b 100644 --- a/test/core/end2end/tests/simple_delayed_request.c +++ b/test/core/end2end/tests/simple_delayed_request.c @@ -62,7 +62,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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; } @@ -77,12 +78,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, @@ -93,8 +91,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; @@ -108,7 +105,7 @@ 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", + c = grpc_channel_create_call(f->client, f->cq, "/foo", "foo.test.google.fr", deadline); GPR_ASSERT(c); @@ -138,10 +135,10 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, 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); + &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; @@ -157,11 +154,9 @@ 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")); @@ -178,8 +173,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 80c092cd35..2246ad540c 100644 --- a/test/core/end2end/tests/simple_request.c +++ b/test/core/end2end/tests/simple_request.c @@ -76,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,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; @@ -116,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); @@ -144,10 +141,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) { 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); + &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; @@ -163,11 +160,9 @@ 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")); @@ -184,8 +179,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) { @@ -209,6 +203,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 67e0730f5e..ff00a5aa6e 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,7 +76,8 @@ static void drain_cq(grpc_completion_queue *cq) { static void shutdown_server(grpc_end2end_test_fixture *f) { if (!f->server) return; - grpc_server_shutdown(f->server); + 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,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; @@ -116,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); @@ -144,10 +141,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) { 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); + &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; @@ -163,11 +160,9 @@ 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")); @@ -184,8 +179,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_10_simple_requests(grpc_end2end_test_config config, int initial_sequence_number) { diff --git a/test/core/fling/server.c b/test/core/fling/server.c index 48304ed8d7..e58d721c86 100644 --- a/test/core/fling/server.c +++ b/test/core/fling/server.c @@ -233,7 +233,8 @@ int main(int argc, char **argv) { while (!shutdown_finished) { if (got_sigint && !shutdown_started) { gpr_log(GPR_INFO, "Shutting down due to SIGINT"); - grpc_server_shutdown(server); + grpc_server_shutdown_and_notify(server, cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)).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..b6e4c0fd0a 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,7 +45,9 @@ #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); @@ -59,47 +62,64 @@ 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; + 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; + 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, + grpc_httpcli_post(&g_context, &g_pollset, &req, "hello", 5, n_seconds_time(15), on_finish, (void *)42); - GPR_ASSERT(gpr_event_wait(&g_done, n_seconds_time(20))); + 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) { @@ -130,12 +150,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..95dca3294a 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); @@ -164,10 +164,10 @@ static void read_and_write_test_write_handler(void *data, 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 +198,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. @@ -222,9 +222,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 +250,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 +283,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 +305,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, @@ -329,8 +327,8 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config, 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 +345,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 +369,11 @@ 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..852e71d479 100644 --- a/test/core/iomgr/endpoint_tests.h +++ b/test/core/iomgr/endpoint_tests.h @@ -52,6 +52,6 @@ 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 */ diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c index 2c8a89e4cd..b1be316a4e 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,8 +270,6 @@ 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 */ 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,32 @@ 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 destroy_change_data(fd_change_data *fdc) { - gpr_mu_destroy(&fdc->mu); - gpr_cv_destroy(&fdc->cv); } 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 +437,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 +446,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 +464,31 @@ 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..1f96aac6a4 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..945f7c1bdc 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,9 +88,13 @@ 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, + grpc_tcp_client_connect(must_succeed, NULL, &g_pollset_set, (struct sockaddr *)&addr, addr_len, gpr_inf_future); /* await the connection */ @@ -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, + 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,48 @@ 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, + 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..23bcd19fef 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,17 @@ 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..1e7af5a339 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,43 @@ 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 +164,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..4e9c787908 100644 --- a/test/core/security/credentials_test.c +++ b/test/core/security/credentials_test.c @@ -328,7 +328,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); } @@ -364,7 +364,7 @@ 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); } @@ -424,7 +424,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); } @@ -506,16 +506,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 +528,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); } @@ -587,16 +587,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 +609,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); } @@ -714,18 +714,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 +744,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 +764,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 +808,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 +839,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..9081e14a9b 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,19 @@ 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/completion_queue_test.c b/test/core/surface/completion_queue_test.c index 9e7b2ea1df..f009b5fa06 100644 --- a/test/core/surface/completion_queue_test.c +++ b/test/core/surface/completion_queue_test.c @@ -245,8 +245,6 @@ static void test_threading(int producers, int consumers) { gpr_log(GPR_INFO, "%s: %d producers, %d consumers", "test_threading", producers, consumers); - grpc_completion_queue_dont_poll_test_only(cc); - /* start all threads: they will wait for phase1 */ for (i = 0; i < producers + consumers; i++) { gpr_thd_id id; diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 24595a820f..507fc7a710 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.IsOk()); } } - 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.IsOk()); @@ -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.IsOk()); @@ -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.IsOk()); } @@ -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.IsOk()); } @@ -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.IsOk()); @@ -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.IsOk()); @@ -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.IsOk()); 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.IsOk()); 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 e86681f75a..7b90fcaec1 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().IsOk()); } -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().IsOk()); @@ -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; } |