diff --git a/CMakeLists.txt b/CMakeLists.txt index 0905ed7fc15..17ed31d4910 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -919,6 +919,7 @@ add_library(end2end_nosec_tests test/core/end2end/tests/cancel_in_a_vacuum.cc test/core/end2end/tests/cancel_with_status.cc test/core/end2end/tests/channelz.cc + test/core/end2end/tests/client_streaming.cc test/core/end2end/tests/compressed_payload.cc test/core/end2end/tests/connectivity.cc test/core/end2end/tests/default_host.cc @@ -1050,6 +1051,7 @@ add_library(end2end_tests test/core/end2end/tests/cancel_in_a_vacuum.cc test/core/end2end/tests/cancel_with_status.cc test/core/end2end/tests/channelz.cc + test/core/end2end/tests/client_streaming.cc test/core/end2end/tests/compressed_payload.cc test/core/end2end/tests/connectivity.cc test/core/end2end/tests/default_host.cc diff --git a/Makefile b/Makefile index 9348affabe3..883745259c5 100644 --- a/Makefile +++ b/Makefile @@ -3290,6 +3290,7 @@ LIBEND2END_NOSEC_TESTS_SRC = \ test/core/end2end/tests/cancel_in_a_vacuum.cc \ test/core/end2end/tests/cancel_with_status.cc \ test/core/end2end/tests/channelz.cc \ + test/core/end2end/tests/client_streaming.cc \ test/core/end2end/tests/compressed_payload.cc \ test/core/end2end/tests/connectivity.cc \ test/core/end2end/tests/default_host.cc \ @@ -3402,6 +3403,7 @@ LIBEND2END_TESTS_SRC = \ test/core/end2end/tests/cancel_in_a_vacuum.cc \ test/core/end2end/tests/cancel_with_status.cc \ test/core/end2end/tests/channelz.cc \ + test/core/end2end/tests/client_streaming.cc \ test/core/end2end/tests/compressed_payload.cc \ test/core/end2end/tests/connectivity.cc \ test/core/end2end/tests/default_host.cc \ diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml index 65467e985d1..ca226638154 100644 --- a/build_autogenerated.yaml +++ b/build_autogenerated.yaml @@ -49,6 +49,7 @@ libs: - test/core/end2end/tests/cancel_in_a_vacuum.cc - test/core/end2end/tests/cancel_with_status.cc - test/core/end2end/tests/channelz.cc + - test/core/end2end/tests/client_streaming.cc - test/core/end2end/tests/compressed_payload.cc - test/core/end2end/tests/connectivity.cc - test/core/end2end/tests/default_host.cc @@ -157,6 +158,7 @@ libs: - test/core/end2end/tests/cancel_in_a_vacuum.cc - test/core/end2end/tests/cancel_with_status.cc - test/core/end2end/tests/channelz.cc + - test/core/end2end/tests/client_streaming.cc - test/core/end2end/tests/compressed_payload.cc - test/core/end2end/tests/connectivity.cc - test/core/end2end/tests/default_host.cc diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 1b97719431e..b2595ac1ce4 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -1487,6 +1487,7 @@ Pod::Spec.new do |s| 'test/core/end2end/tests/cancel_test_helpers.h', 'test/core/end2end/tests/cancel_with_status.cc', 'test/core/end2end/tests/channelz.cc', + 'test/core/end2end/tests/client_streaming.cc', 'test/core/end2end/tests/compressed_payload.cc', 'test/core/end2end/tests/connectivity.cc', 'test/core/end2end/tests/default_host.cc', diff --git a/grpc.gyp b/grpc.gyp index 933251cde76..865c403fde7 100644 --- a/grpc.gyp +++ b/grpc.gyp @@ -198,6 +198,7 @@ 'test/core/end2end/tests/cancel_in_a_vacuum.cc', 'test/core/end2end/tests/cancel_with_status.cc', 'test/core/end2end/tests/channelz.cc', + 'test/core/end2end/tests/client_streaming.cc', 'test/core/end2end/tests/compressed_payload.cc', 'test/core/end2end/tests/connectivity.cc', 'test/core/end2end/tests/default_host.cc', @@ -299,6 +300,7 @@ 'test/core/end2end/tests/cancel_in_a_vacuum.cc', 'test/core/end2end/tests/cancel_with_status.cc', 'test/core/end2end/tests/channelz.cc', + 'test/core/end2end/tests/client_streaming.cc', 'test/core/end2end/tests/compressed_payload.cc', 'test/core/end2end/tests/connectivity.cc', 'test/core/end2end/tests/default_host.cc', diff --git a/src/core/ext/transport/inproc/inproc_transport.cc b/src/core/ext/transport/inproc/inproc_transport.cc index 7b24347c073..5892132418d 100644 --- a/src/core/ext/transport/inproc/inproc_transport.cc +++ b/src/core/ext/transport/inproc/inproc_transport.cc @@ -764,6 +764,7 @@ void op_state_machine_locked(inproc_stream* s, grpc_error* error) { // Nothing further will try to receive from this stream, so finish off // any outstanding send_message op s->send_message_op->payload->send_message.send_message.reset(); + s->send_message_op->payload->send_message.stream_write_closed = true; complete_if_batch_end_locked( s, new_err, s->send_message_op, "op_state_machine scheduling send-message-on-complete"); diff --git a/test/core/end2end/cq_verifier.cc b/test/core/end2end/cq_verifier.cc index 3d45b6d647c..f4e7fe29cf1 100644 --- a/test/core/end2end/cq_verifier.cc +++ b/test/core/end2end/cq_verifier.cc @@ -52,6 +52,7 @@ typedef struct expectation { int line; grpc_completion_type type; void* tag; + bool check_success; int success; } expectation; @@ -220,7 +221,7 @@ static void verify_matches(expectation* e, grpc_event* ev) { GPR_ASSERT(e->type == ev->type); switch (e->type) { case GRPC_OP_COMPLETE: - if (e->success != ev->success) { + if (e->check_success && e->success != ev->success) { gpr_strvec expected; gpr_strvec_init(&expected); expectation_to_strvec(&expected, e); @@ -299,12 +300,14 @@ void cq_verify_empty_timeout(cq_verifier* v, int timeout_sec) { void cq_verify_empty(cq_verifier* v) { cq_verify_empty_timeout(v, 1); } static void add(cq_verifier* v, const char* file, int line, - grpc_completion_type type, void* tag, bool success) { + grpc_completion_type type, void* tag, bool check_success, + bool success) { expectation* e = static_cast(gpr_malloc(sizeof(expectation))); e->type = type; e->file = file; e->line = line; e->tag = tag; + e->check_success = check_success; e->success = success; e->next = v->first_expectation; v->first_expectation = e; @@ -312,5 +315,10 @@ static void add(cq_verifier* v, const char* file, int line, void cq_expect_completion(cq_verifier* v, const char* file, int line, void* tag, bool success) { - add(v, file, line, GRPC_OP_COMPLETE, tag, success); + add(v, file, line, GRPC_OP_COMPLETE, tag, true, success); +} + +void cq_expect_completion_any_status(cq_verifier* v, const char* file, int line, + void* tag) { + add(v, file, line, GRPC_OP_COMPLETE, tag, false, false); } diff --git a/test/core/end2end/cq_verifier.h b/test/core/end2end/cq_verifier.h index 959f849cb11..f5f2a9f7a33 100644 --- a/test/core/end2end/cq_verifier.h +++ b/test/core/end2end/cq_verifier.h @@ -49,8 +49,12 @@ void cq_verify_empty_timeout(cq_verifier* v, int timeout_sec); the event. */ void cq_expect_completion(cq_verifier* v, const char* file, int line, void* tag, bool success); +void cq_expect_completion_any_status(cq_verifier* v, const char* file, int line, + void* tag); #define CQ_EXPECT_COMPLETION(v, tag, success) \ cq_expect_completion(v, __FILE__, __LINE__, tag, success) +#define CQ_EXPECT_COMPLETION_ANY_STATUS(v, tag) \ + cq_expect_completion_any_status(v, __FILE__, __LINE__, tag) int byte_buffer_eq_slice(grpc_byte_buffer* bb, grpc_slice b); int byte_buffer_eq_string(grpc_byte_buffer* byte_buffer, const char* string); diff --git a/test/core/end2end/end2end_nosec_tests.cc b/test/core/end2end/end2end_nosec_tests.cc index e01e6ad0104..f5c9c5ca294 100644 --- a/test/core/end2end/end2end_nosec_tests.cc +++ b/test/core/end2end/end2end_nosec_tests.cc @@ -55,6 +55,8 @@ extern void cancel_with_status(grpc_end2end_test_config config); extern void cancel_with_status_pre_init(void); extern void channelz(grpc_end2end_test_config config); extern void channelz_pre_init(void); +extern void client_streaming(grpc_end2end_test_config config); +extern void client_streaming_pre_init(void); extern void compressed_payload(grpc_end2end_test_config config); extern void compressed_payload_pre_init(void); extern void connectivity(grpc_end2end_test_config config); @@ -200,6 +202,7 @@ void grpc_end2end_tests_pre_init(void) { cancel_in_a_vacuum_pre_init(); cancel_with_status_pre_init(); channelz_pre_init(); + client_streaming_pre_init(); compressed_payload_pre_init(); connectivity_pre_init(); default_host_pre_init(); @@ -286,6 +289,7 @@ void grpc_end2end_tests(int argc, char **argv, cancel_in_a_vacuum(config); cancel_with_status(config); channelz(config); + client_streaming(config); compressed_payload(config); connectivity(config); default_host(config); @@ -406,6 +410,10 @@ void grpc_end2end_tests(int argc, char **argv, channelz(config); continue; } + if (0 == strcmp("client_streaming", argv[i])) { + client_streaming(config); + continue; + } if (0 == strcmp("compressed_payload", argv[i])) { compressed_payload(config); continue; diff --git a/test/core/end2end/end2end_tests.cc b/test/core/end2end/end2end_tests.cc index 76fb046b367..3d81b61eb85 100644 --- a/test/core/end2end/end2end_tests.cc +++ b/test/core/end2end/end2end_tests.cc @@ -57,6 +57,8 @@ extern void cancel_with_status(grpc_end2end_test_config config); extern void cancel_with_status_pre_init(void); extern void channelz(grpc_end2end_test_config config); extern void channelz_pre_init(void); +extern void client_streaming(grpc_end2end_test_config config); +extern void client_streaming_pre_init(void); extern void compressed_payload(grpc_end2end_test_config config); extern void compressed_payload_pre_init(void); extern void connectivity(grpc_end2end_test_config config); @@ -203,6 +205,7 @@ void grpc_end2end_tests_pre_init(void) { cancel_in_a_vacuum_pre_init(); cancel_with_status_pre_init(); channelz_pre_init(); + client_streaming_pre_init(); compressed_payload_pre_init(); connectivity_pre_init(); default_host_pre_init(); @@ -290,6 +293,7 @@ void grpc_end2end_tests(int argc, char **argv, cancel_in_a_vacuum(config); cancel_with_status(config); channelz(config); + client_streaming(config); compressed_payload(config); connectivity(config); default_host(config); @@ -414,6 +418,10 @@ void grpc_end2end_tests(int argc, char **argv, channelz(config); continue; } + if (0 == strcmp("client_streaming", argv[i])) { + client_streaming(config); + continue; + } if (0 == strcmp("compressed_payload", argv[i])) { compressed_payload(config); continue; diff --git a/test/core/end2end/generate_tests.bzl b/test/core/end2end/generate_tests.bzl index 9e93745c7cf..83a2541a14b 100755 --- a/test/core/end2end/generate_tests.bzl +++ b/test/core/end2end/generate_tests.bzl @@ -219,6 +219,7 @@ END2END_TESTS = { "cancel_before_invoke": _test_options(), "cancel_in_a_vacuum": _test_options(), "cancel_with_status": _test_options(), + "client_streaming": _test_options(), "compressed_payload": _test_options(proxyable = False, exclude_inproc = True), "connectivity": _test_options( needs_fullstack = True, diff --git a/test/core/end2end/tests/client_streaming.cc b/test/core/end2end/tests/client_streaming.cc new file mode 100644 index 00000000000..e3f8868ed16 --- /dev/null +++ b/test/core/end2end/tests/client_streaming.cc @@ -0,0 +1,273 @@ +// +// +// Copyright 2020 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// + +#include "test/core/end2end/end2end_tests.h" + +#include +#include + +#include +#include +#include +#include +#include "test/core/end2end/cq_verifier.h" + +static void* tag(intptr_t t) { return (void*)t; } + +static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, + const char* test_name, + grpc_channel_args* client_args, + grpc_channel_args* server_args) { + grpc_end2end_test_fixture f; + gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name); + f = config.create_fixture(client_args, server_args); + config.init_server(&f, server_args); + config.init_client(&f, client_args); + return f; +} + +static gpr_timespec n_seconds_from_now(int n) { + return grpc_timeout_seconds_to_deadline(n); +} + +static gpr_timespec five_seconds_from_now(void) { + return n_seconds_from_now(5); +} + +static void drain_cq(grpc_completion_queue* cq) { + grpc_event ev; + do { + ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr); + } while (ev.type != GRPC_QUEUE_SHUTDOWN); +} + +static void shutdown_server(grpc_end2end_test_fixture* f) { + if (!f->server) return; + grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000)); + GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), + grpc_timeout_seconds_to_deadline(5), + nullptr) + .type == GRPC_OP_COMPLETE); + grpc_server_destroy(f->server); + f->server = nullptr; +} + +static void shutdown_client(grpc_end2end_test_fixture* f) { + if (!f->client) return; + grpc_channel_destroy(f->client); + f->client = nullptr; +} + +static void end_test(grpc_end2end_test_fixture* f) { + shutdown_server(f); + shutdown_client(f); + + grpc_completion_queue_shutdown(f->cq); + drain_cq(f->cq); + grpc_completion_queue_destroy(f->cq); + grpc_completion_queue_destroy(f->shutdown_cq); +} + +// Client streaming test where the client sends a bunch of messages and the +// server reads them. After reading some messages, the server sends the status. +// Client writes fail after that due to the end of stream and the client +// subsequently requests and receives the status. +static void test_client_streaming(grpc_end2end_test_config config, + int messages) { + grpc_end2end_test_fixture f = + begin_test(config, "test_client_streaming", nullptr, nullptr); + grpc_call* c; + grpc_call* s; + cq_verifier* cqv = cq_verifier_create(f.cq); + grpc_op ops[6]; + grpc_op* op; + grpc_metadata_array initial_metadata_recv; + grpc_metadata_array trailing_metadata_recv; + grpc_metadata_array request_metadata_recv; + grpc_call_details call_details; + grpc_status_code status; + grpc_call_error error; + grpc_slice details; + grpc_byte_buffer* request_payload_recv = nullptr; + grpc_byte_buffer* request_payload = nullptr; + int i; + grpc_slice request_payload_slice = + grpc_slice_from_copied_string("hello world"); + + gpr_timespec deadline = five_seconds_from_now(); + c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq, + grpc_slice_from_static_string("/foo"), nullptr, + deadline, nullptr); + GPR_ASSERT(c); + + grpc_metadata_array_init(&initial_metadata_recv); + grpc_metadata_array_init(&trailing_metadata_recv); + grpc_metadata_array_init(&request_metadata_recv); + grpc_call_details_init(&call_details); + + 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 = nullptr; + op++; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(c, ops, static_cast(op - ops), tag(1), + nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + + error = + grpc_server_request_call(f.server, &s, &call_details, + &request_metadata_recv, f.cq, f.cq, tag(100)); + GPR_ASSERT(GRPC_CALL_OK == error); + CQ_EXPECT_COMPLETION(cqv, tag(100), 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 = nullptr; + op++; + error = grpc_call_start_batch(s, ops, static_cast(op - ops), tag(101), + nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + + CQ_EXPECT_COMPLETION(cqv, tag(101), 1); + CQ_EXPECT_COMPLETION(cqv, tag(1), 1); + cq_verify(cqv); + + // Client writes bunch of messages and server reads them + for (i = 0; i < messages; i++) { + request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_MESSAGE; + op->data.send_message.send_message = request_payload; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(c, ops, static_cast(op - ops), + tag(103), nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + grpc_byte_buffer_destroy(request_payload); + + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_RECV_MESSAGE; + op->data.recv_message.recv_message = &request_payload_recv; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(s, ops, static_cast(op - ops), + tag(102), nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + CQ_EXPECT_COMPLETION(cqv, tag(102), 1); + CQ_EXPECT_COMPLETION(cqv, tag(103), 1); + cq_verify(cqv); + GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, "hello world")); + grpc_byte_buffer_destroy(request_payload_recv); + } + + // Server sends status denoting end of stream + memset(ops, 0, sizeof(ops)); + 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; + grpc_slice status_details = grpc_slice_from_static_string("xyz"); + op->data.send_status_from_server.status_details = &status_details; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(s, ops, static_cast(op - ops), tag(104), + nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + CQ_EXPECT_COMPLETION(cqv, tag(104), 1); + cq_verify(cqv); + // Do an empty verify to make sure that the client receives the status + cq_verify_empty(cqv); + + // Client tries sending another message which should fail + request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_MESSAGE; + op->data.send_message.send_message = request_payload; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(c, ops, static_cast(op - ops), tag(103), + nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + grpc_byte_buffer_destroy(request_payload); + CQ_EXPECT_COMPLETION(cqv, tag(103), 0); + cq_verify(cqv); + + // Client sends close and requests status + memset(ops, 0, sizeof(ops)); + op = ops; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; + op->reserved = nullptr; + op++; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; + op->data.recv_status_on_client.status = &status; + op->data.recv_status_on_client.status_details = &details; + op->flags = 0; + op->reserved = nullptr; + op++; + error = grpc_call_start_batch(c, ops, static_cast(op - ops), tag(3), + nullptr); + GPR_ASSERT(GRPC_CALL_OK == error); + CQ_EXPECT_COMPLETION(cqv, tag(3), 1); + cq_verify(cqv); + GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); + GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz")); + + grpc_slice_unref(request_payload_slice); + + grpc_call_unref(c); + grpc_call_unref(s); + + cq_verifier_destroy(cqv); + + grpc_metadata_array_destroy(&initial_metadata_recv); + grpc_metadata_array_destroy(&trailing_metadata_recv); + grpc_metadata_array_destroy(&request_metadata_recv); + grpc_call_details_destroy(&call_details); + grpc_slice_unref(details); + + end_test(&f); + config.tear_down_data(&f); +} + +void client_streaming(grpc_end2end_test_config config) { + for (int i = 0; i < 10; i++) { + test_client_streaming(config, i); + } +} + +void client_streaming_pre_init(void) {} diff --git a/test/core/end2end/tests/streaming_error_response.cc b/test/core/end2end/tests/streaming_error_response.cc index 713975a33f1..ac783611a7b 100644 --- a/test/core/end2end/tests/streaming_error_response.cc +++ b/test/core/end2end/tests/streaming_error_response.cc @@ -88,7 +88,7 @@ static void end_test(grpc_end2end_test_fixture* f) { grpc_completion_queue_destroy(f->shutdown_cq); } -/* Client sends a request with payload, server reads then returns status. */ +// Client sends a request with payload, server reads then returns status. static void test(grpc_end2end_test_config config, bool request_status_early, bool recv_message_separately) { grpc_call* c; @@ -200,8 +200,12 @@ static void test(grpc_end2end_test_config config, bool request_status_early, error = grpc_call_start_batch(s, ops, static_cast(op - ops), tag(103), nullptr); GPR_ASSERT(GRPC_CALL_OK == error); - - CQ_EXPECT_COMPLETION(cqv, tag(103), 1); + // The success of the op depends on whether the payload is written before the + // transport sees the end of stream. If the stream has been write closed + // before the write completes, it would fail, otherwise it would succeed. + // Since this behavior is dependent on the transport implementation, we allow + // any success status with this op. + CQ_EXPECT_COMPLETION_ANY_STATUS(cqv, tag(103)); if (!request_status_early) { memset(ops, 0, sizeof(ops));