From 5ba5c49bfa80c5bb535873ed090247c2bde1c0be Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 4 Jan 2017 09:22:47 -0800 Subject: [PATCH] Add test of deadline expiry with max concurrent streams --- .../end2end/tests/max_concurrent_streams.c | 193 ++++++++++++++++++ 1 file changed, 193 insertions(+) diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c index 9338bc5f0d7..e6502a7b0cd 100644 --- a/test/core/end2end/tests/max_concurrent_streams.c +++ b/test/core/end2end/tests/max_concurrent_streams.c @@ -444,7 +444,200 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { config.tear_down_data(&f); } +static void test_max_concurrent_streams_with_timeout( + grpc_end2end_test_config config) { + grpc_end2end_test_fixture f; + grpc_arg server_arg; + grpc_channel_args server_args; + grpc_call *c1; + grpc_call *c2; + grpc_call *s1; + grpc_call *s2; + cq_verifier *cqv; + grpc_call_details call_details; + grpc_metadata_array request_metadata_recv; + grpc_metadata_array initial_metadata_recv1; + grpc_metadata_array trailing_metadata_recv1; + grpc_metadata_array initial_metadata_recv2; + grpc_metadata_array trailing_metadata_recv2; + grpc_status_code status1; + grpc_call_error error; + char *details1 = NULL; + size_t details_capacity1 = 0; + grpc_status_code status2; + char *details2 = NULL; + size_t details_capacity2 = 0; + grpc_op ops[6]; + grpc_op *op; + int was_cancelled; + + server_arg.key = GRPC_ARG_MAX_CONCURRENT_STREAMS; + server_arg.type = GRPC_ARG_INTEGER; + server_arg.value.integer = 1; + + server_args.num_args = 1; + server_args.args = &server_arg; + + f = begin_test(config, "test_max_concurrent_streams_with_timeout", NULL, + &server_args); + cqv = cq_verifier_create(f.cq); + + grpc_metadata_array_init(&request_metadata_recv); + grpc_metadata_array_init(&initial_metadata_recv1); + grpc_metadata_array_init(&trailing_metadata_recv1); + grpc_metadata_array_init(&initial_metadata_recv2); + grpc_metadata_array_init(&trailing_metadata_recv2); + grpc_call_details_init(&call_details); + + /* perform a ping-pong to ensure that settings have had a chance to round + trip */ + simple_request_body(config, f); + /* perform another one to make sure that the one stream case still works */ + simple_request_body(config, f); + + /* start two requests - ensuring that the second is not accepted until + the first completes */ + c1 = grpc_channel_create_call( + f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, "/alpha", + get_host_override_string("foo.test.google.fr:1234", config), + n_seconds_time(3), NULL); + GPR_ASSERT(c1); + c2 = grpc_channel_create_call( + f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, "/beta", + get_host_override_string("foo.test.google.fr:1234", config), + n_seconds_time(1000), NULL); + GPR_ASSERT(c2); + + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s1, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(101))); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; + op->reserved = NULL; + op++; + error = grpc_call_start_batch(c1, ops, (size_t)(op - ops), tag(301), NULL); + GPR_ASSERT(GRPC_CALL_OK == error); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv1; + op->data.recv_status_on_client.status = &status1; + op->data.recv_status_on_client.status_details = &details1; + op->data.recv_status_on_client.status_details_capacity = &details_capacity1; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata = &initial_metadata_recv1; + op->flags = 0; + op->reserved = NULL; + op++; + error = grpc_call_start_batch(c1, ops, (size_t)(op - ops), tag(302), NULL); + GPR_ASSERT(GRPC_CALL_OK == error); + + CQ_EXPECT_COMPLETION(cqv, tag(101), 1); + CQ_EXPECT_COMPLETION(cqv, tag(301), 1); + cq_verify(cqv); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; + op->reserved = NULL; + op++; + error = grpc_call_start_batch(c2, ops, (size_t)(op - ops), tag(401), NULL); + GPR_ASSERT(GRPC_CALL_OK == error); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv2; + op->data.recv_status_on_client.status = &status2; + op->data.recv_status_on_client.status_details = &details2; + op->data.recv_status_on_client.status_details_capacity = &details_capacity2; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata = &initial_metadata_recv1; + op->flags = 0; + op->reserved = NULL; + op++; + error = grpc_call_start_batch(c2, ops, (size_t)(op - ops), tag(402), NULL); + GPR_ASSERT(GRPC_CALL_OK == error); + + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f.server, &s2, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(201))); + + CQ_EXPECT_COMPLETION(cqv, tag(302), 1); + /* first request is finished, we should be able to start the second */ + CQ_EXPECT_COMPLETION(cqv, tag(401), 1); + CQ_EXPECT_COMPLETION(cqv, tag(201), 1); + cq_verify(cqv); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; + op->data.recv_close_on_server.cancelled = &was_cancelled; + op->flags = 0; + op->reserved = NULL; + op++; + op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; + op->data.send_status_from_server.trailing_metadata_count = 0; + op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED; + op->data.send_status_from_server.status_details = "xyz"; + op->flags = 0; + op->reserved = NULL; + op++; + error = grpc_call_start_batch(s2, ops, (size_t)(op - ops), tag(202), NULL); + GPR_ASSERT(GRPC_CALL_OK == error); + + CQ_EXPECT_COMPLETION(cqv, tag(402), 1); + CQ_EXPECT_COMPLETION(cqv, tag(202), 1); + cq_verify(cqv); + + cq_verifier_destroy(cqv); + + grpc_call_destroy(c1); + grpc_call_destroy(s1); + grpc_call_destroy(c2); + grpc_call_destroy(s2); + + gpr_free(details1); + gpr_free(details2); + grpc_metadata_array_destroy(&initial_metadata_recv1); + grpc_metadata_array_destroy(&trailing_metadata_recv1); + grpc_metadata_array_destroy(&initial_metadata_recv2); + grpc_metadata_array_destroy(&trailing_metadata_recv2); + grpc_metadata_array_destroy(&request_metadata_recv); + grpc_call_details_destroy(&call_details); + + end_test(&f); + config.tear_down_data(&f); +} + void max_concurrent_streams(grpc_end2end_test_config config) { + test_max_concurrent_streams_with_timeout(config); test_max_concurrent_streams(config); }