diff options
author | 2015-05-05 08:44:47 -0700 | |
---|---|---|
committer | 2015-05-05 08:44:47 -0700 | |
commit | 5fdcbb7fbebe6efd353494a8f316216f13f7cf13 (patch) | |
tree | 50ca95b4a61bf94954e1c1db0f138cc29b47487a /test | |
parent | 8e7fe9b9f01f88e322b97bdd5ed399395e563cde (diff) | |
parent | d53d87cbd3a93fa90ac1363fd5dc3b27dbb31cae (diff) |
Merge github.com:grpc/grpc into bye-bye-completion-queue-pie
Diffstat (limited to 'test')
-rw-r--r-- | test/build/systemtap.c | 4 | ||||
-rwxr-xr-x | test/core/end2end/gen_build_json.py | 2 | ||||
-rw-r--r-- | test/core/end2end/tests/max_message_length.c | 210 | ||||
-rw-r--r-- | test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c | 223 | ||||
-rw-r--r-- | test/cpp/end2end/end2end_test.cc | 24 | ||||
-rw-r--r-- | test/cpp/qps/client_async.cc | 133 |
6 files changed, 503 insertions, 93 deletions
diff --git a/test/build/systemtap.c b/test/build/systemtap.c index 66ff38ebd6..1997d5f17b 100644 --- a/test/build/systemtap.c +++ b/test/build/systemtap.c @@ -37,6 +37,4 @@ #error "_SYS_SDT_H not defined, despite <sys/sdt.h> being present." #endif -int main() { - return 0; -} +int main() { return 0; } diff --git a/test/core/end2end/gen_build_json.py b/test/core/end2end/gen_build_json.py index ac7c194d8c..1d981cd3ad 100755 --- a/test/core/end2end/gen_build_json.py +++ b/test/core/end2end/gen_build_json.py @@ -62,6 +62,7 @@ END2END_TESTS = { 'graceful_server_shutdown': True, 'invoke_large_request': False, 'max_concurrent_streams': True, + 'max_message_length': True, 'no_op': True, 'ping_pong_streaming': True, 'request_response_with_binary_metadata_and_payload': True, @@ -71,6 +72,7 @@ END2END_TESTS = { 'request_with_payload': True, 'simple_delayed_request': True, 'simple_request': True, + 'simple_request_with_high_initial_sequence_number': True, 'registered_call': True, } diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c new file mode 100644 index 0000000000..6291f773b3 --- /dev/null +++ b/test/core/end2end/tests/max_message_length.c @@ -0,0 +1,210 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "test/core/end2end/end2end_tests.h" + +#include <stdio.h> +#include <string.h> +#include <unistd.h> + +#include <grpc/byte_buffer.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/time.h> +#include <grpc/support/useful.h> +#include "test/core/end2end/cq_verifier.h" + +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, + grpc_channel_args *server_args) { + grpc_end2end_test_fixture f; + gpr_log(GPR_INFO, "%s/%s", test_name, config.name); + f = config.create_fixture(client_args, server_args); + config.init_client(&f, client_args); + config.init_server(&f, server_args); + return f; +} + +static gpr_timespec n_seconds_time(int n) { + return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n); +} + +static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); } + +static void drain_cq(grpc_completion_queue *cq) { + grpc_event *ev; + grpc_completion_type type; + do { + ev = grpc_completion_queue_next(cq, five_seconds_time()); + GPR_ASSERT(ev); + type = ev->type; + grpc_event_finish(ev); + } while (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); + f->client = NULL; +} + +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); +} + +static void test_max_message_length(grpc_end2end_test_config config) { + grpc_end2end_test_fixture f; + grpc_arg server_arg; + grpc_channel_args server_args; + grpc_call *c; + grpc_call *s; + cq_verifier *v_client; + cq_verifier *v_server; + grpc_op ops[6]; + grpc_op *op; + gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world"); + grpc_byte_buffer *request_payload = + grpc_byte_buffer_create(&request_payload_slice, 1); + 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; + char *details = NULL; + size_t details_capacity = 0; + int was_cancelled = 2; + + server_arg.key = GRPC_ARG_MAX_MESSAGE_LENGTH; + server_arg.type = GRPC_ARG_INTEGER; + server_arg.value.integer = 5; + + server_args.num_args = 1; + server_args.args = &server_arg; + + f = begin_test(config, __FUNCTION__, NULL, &server_args); + v_client = cq_verifier_create(f.client_cq); + v_server = cq_verifier_create(f.server_cq); + + c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + "foo.test.google.fr:1234", gpr_inf_future); + 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); + + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op++; + op->op = GRPC_OP_SEND_MESSAGE; + op->data.send_message = request_payload; + op++; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op++; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata = &initial_metadata_recv; + op++; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; + op->data.recv_status_on_client.status = &status; + op->data.recv_status_on_client.status_details = &details; + op->data.recv_status_on_client.status_details_capacity = &details_capacity; + op++; + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); + + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, + &call_details, + &request_metadata_recv, + f.server_cq, tag(101))); + cq_expect_completion(v_server, tag(101), GRPC_OP_OK); + cq_verify(v_server); + + op = ops; + op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; + op->data.recv_close_on_server.cancelled = &was_cancelled; + op++; + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); + + cq_expect_completion(v_server, tag(102), GRPC_OP_OK); + cq_verify(v_server); + + cq_expect_completion(v_client, tag(1), GRPC_OP_OK); + cq_verify(v_client); + + GPR_ASSERT(status == GRPC_STATUS_CANCELLED); + GPR_ASSERT(0 == strcmp(details, "Cancelled")); + GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); + GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); + GPR_ASSERT(was_cancelled == 1); + + gpr_free(details); + 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_call_destroy(c); + grpc_call_destroy(s); + + cq_verifier_destroy(v_client); + cq_verifier_destroy(v_server); + + end_test(&f); + config.tear_down_data(&f); +} + +void grpc_end2end_tests(grpc_end2end_test_config config) { + test_max_message_length(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 new file mode 100644 index 0000000000..538291a5f2 --- /dev/null +++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c @@ -0,0 +1,223 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "test/core/end2end/end2end_tests.h" + +#include <stdio.h> +#include <string.h> +#include <unistd.h> + +#include "src/core/support/string.h" +#include <grpc/byte_buffer.h> +#include <grpc/grpc.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/time.h> +#include <grpc/support/useful.h> +#include "test/core/end2end/cq_verifier.h" + +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, + grpc_channel_args *server_args) { + grpc_end2end_test_fixture f; + gpr_log(GPR_INFO, "%s/%s", test_name, config.name); + f = config.create_fixture(client_args, server_args); + config.init_client(&f, client_args); + config.init_server(&f, server_args); + return f; +} + +static gpr_timespec n_seconds_time(int n) { + return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n); +} + +static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); } + +static void drain_cq(grpc_completion_queue *cq) { + grpc_event *ev; + grpc_completion_type type; + do { + ev = grpc_completion_queue_next(cq, five_seconds_time()); + GPR_ASSERT(ev); + type = ev->type; + grpc_event_finish(ev); + } while (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); + f->client = NULL; +} + +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); +} + +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); + 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; + char *details = NULL; + size_t details_capacity = 0; + int was_cancelled = 2; + + c = grpc_channel_create_call(f.client, f.client_cq, "/foo", + "foo.test.google.fr:1234", deadline); + 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); + + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + op++; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op++; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata = &initial_metadata_recv; + op++; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; + op->data.recv_status_on_client.status = &status; + op->data.recv_status_on_client.status_details = &details; + op->data.recv_status_on_client.status_details_capacity = &details_capacity; + op++; + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); + + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s, + &call_details, + &request_metadata_recv, + f.server_cq, tag(101))); + cq_expect_completion(v_server, tag(101), GRPC_OP_OK); + cq_verify(v_server); + + op = ops; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->data.send_initial_metadata.count = 0; + 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++; + op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; + op->data.recv_close_on_server.cancelled = &was_cancelled; + op++; + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); + + cq_expect_completion(v_server, tag(102), GRPC_OP_OK); + cq_verify(v_server); + + cq_expect_completion(v_client, tag(1), GRPC_OP_OK); + cq_verify(v_client); + + GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED); + GPR_ASSERT(0 == strcmp(details, "xyz")); + GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); + GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); + GPR_ASSERT(was_cancelled == 0); + + gpr_free(details); + 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_call_destroy(c); + grpc_call_destroy(s); + + cq_verifier_destroy(v_client); + cq_verifier_destroy(v_server); +} + +static void test_invoke_10_simple_requests(grpc_end2end_test_config config, int initial_sequence_number) { + int i; + grpc_end2end_test_fixture f; + grpc_arg client_arg; + grpc_channel_args client_args; + + client_arg.type = GRPC_ARG_INTEGER; + client_arg.key = GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER; + client_arg.value.integer = initial_sequence_number; + + client_args.num_args = 1; + client_args.args = &client_arg; + + f = begin_test(config, __FUNCTION__, &client_args, NULL); + for (i = 0; i < 10; i++) { + simple_request_body(f); + gpr_log(GPR_INFO, "Passed simple request %d", i); + } + end_test(&f); + config.tear_down_data(&f); +} + +void grpc_end2end_tests(grpc_end2end_test_config config) { + test_invoke_10_simple_requests(config, 16777213); + if (config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION) { + test_invoke_10_simple_requests(config, 2147483645); + } +} diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 93d7ace9a2..0945ed269d 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -172,7 +172,7 @@ class TestServiceImplDupPkg class End2endTest : public ::testing::Test { protected: - End2endTest() : thread_pool_(2) {} + End2endTest() : kMaxMessageSize_(8192), thread_pool_(2) {} void SetUp() GRPC_OVERRIDE { int port = grpc_pick_unused_port_or_die(); @@ -182,6 +182,8 @@ class End2endTest : public ::testing::Test { builder.AddListeningPort(server_address_.str(), InsecureServerCredentials()); builder.RegisterService(&service_); + builder.SetMaxMessageSize( + kMaxMessageSize_); // For testing max message size. builder.RegisterService(&dup_pkg_service_); builder.SetThreadPool(&thread_pool_); server_ = builder.BuildAndStart(); @@ -198,6 +200,7 @@ class End2endTest : public ::testing::Test { std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_; std::unique_ptr<Server> server_; std::ostringstream server_address_; + const int kMaxMessageSize_; TestServiceImpl service_; TestServiceImplDupPkg dup_pkg_service_; ThreadPool thread_pool_; @@ -426,8 +429,7 @@ TEST_F(End2endTest, DiffPackageServices) { // rpc and stream should fail on bad credentials. TEST_F(End2endTest, BadCredentials) { - std::unique_ptr<Credentials> bad_creds = - ServiceAccountCredentials("", "", 1); + std::unique_ptr<Credentials> bad_creds = ServiceAccountCredentials("", "", 1); EXPECT_EQ(nullptr, bad_creds.get()); std::shared_ptr<ChannelInterface> channel = CreateChannel(server_address_.str(), bad_creds, ChannelArguments()); @@ -501,14 +503,13 @@ TEST_F(End2endTest, ClientCancelsRequestStream) { auto stream = stub_->RequestStream(&context, &response); EXPECT_TRUE(stream->Write(request)); EXPECT_TRUE(stream->Write(request)); - + context.TryCancel(); Status s = stream->Finish(); EXPECT_EQ(grpc::StatusCode::CANCELLED, s.code()); - - EXPECT_EQ(response.message(), ""); + EXPECT_EQ(response.message(), ""); } // Client cancels server stream after sending some messages @@ -588,6 +589,17 @@ TEST_F(End2endTest, ThreadStress) { } } +TEST_F(End2endTest, RpcMaxMessageSize) { + ResetStub(); + EchoRequest request; + EchoResponse response; + request.set_message(string(kMaxMessageSize_ * 2, 'a')); + + ClientContext context; + Status s = stub_->Echo(&context, request, &response); + EXPECT_FALSE(s.IsOk()); +} + } // namespace testing } // namespace grpc diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc index 0a6d9beeca..0aec1b1a57 100644 --- a/test/cpp/qps/client_async.cc +++ b/test/cpp/qps/client_async.cc @@ -130,39 +130,26 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { response_reader_; }; -class AsyncUnaryClient GRPC_FINAL : public Client { +class AsyncClient : public Client { public: - explicit AsyncUnaryClient(const ClientConfig& config) : Client(config) { + explicit AsyncClient(const ClientConfig& config, + std::function<void(CompletionQueue*, TestService::Stub*, + const SimpleRequest&)> setup_ctx) : + Client(config) { for (int i = 0; i < config.async_client_threads(); i++) { cli_cqs_.emplace_back(new CompletionQueue); } - - auto check_done = [](grpc::Status s, SimpleResponse* response) {}; - int t = 0; for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) { for (auto channel = channels_.begin(); channel != channels_.end(); channel++) { auto* cq = cli_cqs_[t].get(); t = (t + 1) % cli_cqs_.size(); - auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx, - const SimpleRequest& request, void* tag) { - return stub->AsyncUnaryCall(ctx, request, cq, tag); - }; - - TestService::Stub* stub = channel->get_stub(); - const SimpleRequest& request = request_; - new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( - stub, request, start_req, check_done); + setup_ctx(cq, channel->get_stub(), request_); } } - - StartThreads(config.async_client_threads()); } - - ~AsyncUnaryClient() GRPC_OVERRIDE { - EndThreads(); - + virtual ~AsyncClient() { for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) { (*cq)->Shutdown(); void* got_tag; @@ -173,10 +160,13 @@ class AsyncUnaryClient GRPC_FINAL : public Client { } } - bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE { + bool ThreadFunc(Histogram* histogram, size_t thread_idx) + GRPC_OVERRIDE GRPC_FINAL { void* got_tag; bool ok; - switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, std::chrono::system_clock::now() + std::chrono::seconds(1))) { + switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, + std::chrono::system_clock::now() + + std::chrono::seconds(1))) { case CompletionQueue::SHUTDOWN: return false; case CompletionQueue::TIMEOUT: return true; case CompletionQueue::GOT_EVENT: break; @@ -192,10 +182,30 @@ class AsyncUnaryClient GRPC_FINAL : public Client { return true; } - + private: std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; }; +class AsyncUnaryClient GRPC_FINAL : public AsyncClient { + public: + explicit AsyncUnaryClient(const ClientConfig& config) : + AsyncClient(config, SetupCtx) { + StartThreads(config.async_client_threads()); + } + ~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); } +private: + static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub, + const SimpleRequest& req) { + auto check_done = [](grpc::Status s, SimpleResponse* response) {}; + auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx, + const SimpleRequest& request, void* tag) { + return stub->AsyncUnaryCall(ctx, request, cq, tag); + }; + new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( + stub, req, start_req, check_done); + } +}; + template <class RequestType, class ResponseType> class ClientRpcContextStreamingImpl : public ClientRpcContext { public: @@ -241,7 +251,7 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext { return(false); } next_state_ = &ClientRpcContextStreamingImpl::ReadDone; - stream_->Read(&response_, ClientRpcContext::tag(this)); + stream_->Read(&response_, ClientRpcContext::tag(this)); return true; } bool ReadDone(bool ok, Histogram *hist) { @@ -263,71 +273,26 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext { stream_; }; -class AsyncStreamingClient GRPC_FINAL : public Client { +class AsyncStreamingClient GRPC_FINAL : public AsyncClient { public: - explicit AsyncStreamingClient(const ClientConfig &config) : Client(config) { - for (int i = 0; i < config.async_client_threads(); i++) { - cli_cqs_.emplace_back(new CompletionQueue); - } - - auto check_done = [](grpc::Status s, SimpleResponse* response) {}; - - int t = 0; - for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) { - for (auto channel = channels_.begin(); channel != channels_.end(); - channel++) { - auto* cq = cli_cqs_[t].get(); - t = (t + 1) % cli_cqs_.size(); - auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx, - void *tag) { - auto stream = stub->AsyncStreamingCall(ctx, cq, tag); - return stream; - }; - - TestService::Stub *stub = channel->get_stub(); - const SimpleRequest &request = request_; - new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( - stub, request, start_req, check_done); - } - } - + explicit AsyncStreamingClient(const ClientConfig &config) : + AsyncClient(config, SetupCtx) { StartThreads(config.async_client_threads()); } - ~AsyncStreamingClient() GRPC_OVERRIDE { - EndThreads(); - - for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) { - (*cq)->Shutdown(); - void *got_tag; - bool ok; - while ((*cq)->Next(&got_tag, &ok)) { - delete ClientRpcContext::detag(got_tag); - } - } - } - - bool ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE { - void *got_tag; - bool ok; - switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, std::chrono::system_clock::now() + std::chrono::seconds(1))) { - case CompletionQueue::SHUTDOWN: return false; - case CompletionQueue::TIMEOUT: return true; - case CompletionQueue::GOT_EVENT: break; - } - - ClientRpcContext *ctx = ClientRpcContext::detag(got_tag); - if (ctx->RunNextState(ok, histogram) == false) { - // call the callback and then delete it - ctx->RunNextState(ok, histogram); - ctx->StartNewClone(); - delete ctx; - } - - return true; + ~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); } +private: + static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub, + const SimpleRequest& req) { + auto check_done = [](grpc::Status s, SimpleResponse* response) {}; + auto start_req = [cq](TestService::Stub *stub, grpc::ClientContext *ctx, + void *tag) { + auto stream = stub->AsyncStreamingCall(ctx, cq, tag); + return stream; + }; + new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( + stub, req, start_req, check_done); } - - std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; }; std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig& args) { |