diff options
Diffstat (limited to 'test/cpp')
27 files changed, 1081 insertions, 238 deletions
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 507fc7a710..117d8bb9fa 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -167,7 +167,7 @@ class AsyncEnd2endTest : public ::testing::Test { Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } } @@ -227,7 +227,7 @@ TEST_F(AsyncEnd2endTest, AsyncNextRpc) { 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()); + EXPECT_TRUE(recv_status.ok()); } // Two pings and a final pong. @@ -280,7 +280,7 @@ TEST_F(AsyncEnd2endTest, SimpleClientStreaming) { Verifier().Expect(10, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } // One ping, two pongs. @@ -330,7 +330,7 @@ TEST_F(AsyncEnd2endTest, SimpleServerStreaming) { cli_stream->Finish(&recv_status, tag(9)); Verifier().Expect(9, true).Verify(cq_.get()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } // One ping, one pong. @@ -382,7 +382,7 @@ TEST_F(AsyncEnd2endTest, SimpleBidiStreaming) { cli_stream->Finish(&recv_status, tag(10)); Verifier().Expect(10, true).Verify(cq_.get()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } // Metadata tests @@ -426,7 +426,7 @@ TEST_F(AsyncEnd2endTest, ClientInitialMetadataRpc) { Verifier().Expect(4, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } TEST_F(AsyncEnd2endTest, ServerInitialMetadataRpc) { @@ -473,7 +473,7 @@ TEST_F(AsyncEnd2endTest, ServerInitialMetadataRpc) { Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } TEST_F(AsyncEnd2endTest, ServerTrailingMetadataRpc) { @@ -513,7 +513,7 @@ TEST_F(AsyncEnd2endTest, ServerTrailingMetadataRpc) { response_reader->Finish(&recv_response, &recv_status, tag(5)); Verifier().Expect(5, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata(); EXPECT_EQ(meta1.second, server_trailing_metadata.find(meta1.first)->second); EXPECT_EQ(meta2.second, server_trailing_metadata.find(meta2.first)->second); @@ -586,7 +586,7 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { response_reader->Finish(&recv_response, &recv_status, tag(6)); Verifier().Expect(6, true).Verify(cq_.get()); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata(); EXPECT_EQ(meta5.second, server_trailing_metadata.find(meta5.first)->second); EXPECT_EQ(meta6.second, server_trailing_metadata.find(meta6.first)->second); diff --git a/test/cpp/end2end/client_crash_test.cc b/test/cpp/end2end/client_crash_test.cc index 7b90fcaec1..7876e8dee3 100644 --- a/test/cpp/end2end/client_crash_test.cc +++ b/test/cpp/end2end/client_crash_test.cc @@ -118,7 +118,7 @@ TEST_F(CrashTest, KillBeforeWrite) { // But the read will definitely fail EXPECT_FALSE(stream->Read(&response)); - EXPECT_FALSE(stream->Finish().IsOk()); + EXPECT_FALSE(stream->Finish().ok()); } TEST_F(CrashTest, KillAfterWrite) { @@ -142,7 +142,7 @@ TEST_F(CrashTest, KillAfterWrite) { EXPECT_FALSE(stream->Read(&response)); - EXPECT_FALSE(stream->Finish().IsOk()); + EXPECT_FALSE(stream->Finish().ok()); } } // namespace diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index f48cf953a4..45ba8b0878 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -101,13 +101,13 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { gpr_now(), gpr_time_from_micros(request->param().client_cancel_after_us()))); } - return Status::Cancelled; + return Status::CANCELLED; } else if (request->has_param() && request->param().server_cancel_after_us()) { gpr_sleep_until(gpr_time_add( gpr_now(), gpr_time_from_micros(request->param().server_cancel_after_us()))); - return Status::Cancelled; + return Status::CANCELLED; } else { EXPECT_FALSE(context->IsCancelled()); } @@ -232,7 +232,7 @@ static void SendRpc(grpc::cpp::test::util::TestService::Stub* stub, ClientContext context; Status s = stub->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } } @@ -265,7 +265,7 @@ TEST_F(End2endTest, RpcDeadlineExpires) { std::chrono::system_clock::now() + std::chrono::microseconds(10); context.set_deadline(deadline); Status s = stub_->Echo(&context, request, &response); - EXPECT_EQ(StatusCode::DEADLINE_EXCEEDED, s.code()); + EXPECT_EQ(StatusCode::DEADLINE_EXCEEDED, s.error_code()); } // Set a long but finite deadline. @@ -281,7 +281,7 @@ TEST_F(End2endTest, RpcLongDeadline) { context.set_deadline(deadline); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } // Ask server to echo back the deadline it sees. @@ -298,7 +298,7 @@ TEST_F(End2endTest, EchoDeadline) { context.set_deadline(deadline); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); gpr_timespec sent_deadline; Timepoint2Timespec(deadline, &sent_deadline); // Allow 1 second error. @@ -317,7 +317,7 @@ TEST_F(End2endTest, EchoDeadlineForNoDeadlineRpc) { ClientContext context; Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); EXPECT_EQ(response.param().request_deadline(), gpr_inf_future.tv_sec); } @@ -329,9 +329,9 @@ TEST_F(End2endTest, UnimplementedRpc) { ClientContext context; Status s = stub_->Unimplemented(&context, request, &response); - EXPECT_FALSE(s.IsOk()); - EXPECT_EQ(s.code(), grpc::StatusCode::UNIMPLEMENTED); - EXPECT_EQ(s.details(), ""); + EXPECT_FALSE(s.ok()); + EXPECT_EQ(s.error_code(), grpc::StatusCode::UNIMPLEMENTED); + EXPECT_EQ(s.error_message(), ""); EXPECT_EQ(response.message(), ""); } @@ -347,7 +347,7 @@ TEST_F(End2endTest, RequestStreamOneRequest) { stream->WritesDone(); Status s = stream->Finish(); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } TEST_F(End2endTest, RequestStreamTwoRequests) { @@ -363,7 +363,7 @@ TEST_F(End2endTest, RequestStreamTwoRequests) { stream->WritesDone(); Status s = stream->Finish(); EXPECT_EQ(response.message(), "hellohello"); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } TEST_F(End2endTest, ResponseStream) { @@ -383,7 +383,7 @@ TEST_F(End2endTest, ResponseStream) { EXPECT_FALSE(stream->Read(&response)); Status s = stream->Finish(); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } TEST_F(End2endTest, BidiStream) { @@ -414,7 +414,7 @@ TEST_F(End2endTest, BidiStream) { EXPECT_FALSE(stream->Read(&response)); Status s = stream->Finish(); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } // Talk to the two services with the same name but different package names. @@ -433,7 +433,7 @@ TEST_F(End2endTest, DiffPackageServices) { ClientContext context; Status s = stub->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); std::unique_ptr<grpc::cpp::test::util::duplicate::TestService::Stub> dup_pkg_stub( @@ -441,7 +441,7 @@ TEST_F(End2endTest, DiffPackageServices) { ClientContext context2; s = dup_pkg_stub->Echo(&context2, request, &response); EXPECT_EQ("no package", response.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } // rpc and stream should fail on bad credentials. @@ -459,16 +459,16 @@ TEST_F(End2endTest, BadCredentials) { Status s = stub->Echo(&context, request, &response); EXPECT_EQ("", response.message()); - EXPECT_FALSE(s.IsOk()); - EXPECT_EQ(StatusCode::UNKNOWN, s.code()); - EXPECT_EQ("Rpc sent on a lame channel.", s.details()); + EXPECT_FALSE(s.ok()); + EXPECT_EQ(StatusCode::UNKNOWN, s.error_code()); + EXPECT_EQ("Rpc sent on a lame channel.", s.error_message()); ClientContext context2; auto stream = stub->BidiStream(&context2); s = stream->Finish(); - EXPECT_FALSE(s.IsOk()); - EXPECT_EQ(StatusCode::UNKNOWN, s.code()); - EXPECT_EQ("Rpc sent on a lame channel.", s.details()); + EXPECT_FALSE(s.ok()); + EXPECT_EQ(StatusCode::UNKNOWN, s.error_code()); + EXPECT_EQ("Rpc sent on a lame channel.", s.error_message()); } void CancelRpc(ClientContext* context, int delay_us, TestServiceImpl* service) { @@ -491,8 +491,8 @@ TEST_F(End2endTest, ClientCancelsRpc) { std::thread cancel_thread(CancelRpc, &context, kCancelDelayUs, &service_); Status s = stub_->Echo(&context, request, &response); cancel_thread.join(); - EXPECT_EQ(StatusCode::CANCELLED, s.code()); - EXPECT_EQ(s.details(), "Cancelled"); + EXPECT_EQ(StatusCode::CANCELLED, s.error_code()); + EXPECT_EQ(s.error_message(), "Cancelled"); } // Server cancels rpc after 1ms @@ -505,8 +505,8 @@ TEST_F(End2endTest, ServerCancelsRpc) { ClientContext context; Status s = stub_->Echo(&context, request, &response); - EXPECT_EQ(StatusCode::CANCELLED, s.code()); - EXPECT_TRUE(s.details().empty()); + EXPECT_EQ(StatusCode::CANCELLED, s.error_code()); + EXPECT_TRUE(s.error_message().empty()); } // Client cancels request stream after sending two messages @@ -524,7 +524,7 @@ TEST_F(End2endTest, ClientCancelsRequestStream) { context.TryCancel(); Status s = stream->Finish(); - EXPECT_EQ(grpc::StatusCode::CANCELLED, s.code()); + EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); EXPECT_EQ(response.message(), ""); } @@ -558,7 +558,7 @@ TEST_F(End2endTest, ClientCancelsResponseStream) { Status s = stream->Finish(); // The final status could be either of CANCELLED or OK depending on // who won the race. - EXPECT_GE(grpc::StatusCode::CANCELLED, s.code()); + EXPECT_GE(grpc::StatusCode::CANCELLED, s.error_code()); } // Client cancels bidi stream after sending some messages @@ -591,7 +591,7 @@ TEST_F(End2endTest, ClientCancelsBidi) { } Status s = stream->Finish(); - EXPECT_EQ(grpc::StatusCode::CANCELLED, s.code()); + EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); } TEST_F(End2endTest, RpcMaxMessageSize) { @@ -602,7 +602,7 @@ TEST_F(End2endTest, RpcMaxMessageSize) { ClientContext context; Status s = stub_->Echo(&context, request, &response); - EXPECT_FALSE(s.IsOk()); + EXPECT_FALSE(s.ok()); } bool MetadataContains(const std::multimap<grpc::string, grpc::string>& metadata, @@ -632,7 +632,7 @@ TEST_F(End2endTest, SetPerCallCredentials) { Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(request.message(), response.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); EXPECT_TRUE(MetadataContains(context.GetServerTrailingMetadata(), GRPC_IAM_AUTHORIZATION_TOKEN_METADATA_KEY, "fake_token")); @@ -652,8 +652,8 @@ TEST_F(End2endTest, InsecurePerCallCredentials) { request.mutable_param()->set_echo_metadata(true); Status s = stub_->Echo(&context, request, &response); - EXPECT_EQ(StatusCode::CANCELLED, s.code()); - EXPECT_EQ("Failed to set credentials to rpc.", s.details()); + EXPECT_EQ(StatusCode::CANCELLED, s.error_code()); + EXPECT_EQ("Failed to set credentials to rpc.", s.error_message()); } TEST_F(End2endTest, OverridePerCallCredentials) { @@ -684,7 +684,7 @@ TEST_F(End2endTest, OverridePerCallCredentials) { GRPC_IAM_AUTHORITY_SELECTOR_METADATA_KEY, "fake_selector1")); EXPECT_EQ(request.message(), response.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } } // namespace testing diff --git a/test/cpp/end2end/generic_end2end_test.cc b/test/cpp/end2end/generic_end2end_test.cc index 80e43fd854..7132b6b1f1 100644 --- a/test/cpp/end2end/generic_end2end_test.cc +++ b/test/cpp/end2end/generic_end2end_test.cc @@ -190,7 +190,7 @@ class GenericEnd2endTest : public ::testing::Test { client_ok(9); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } } @@ -273,7 +273,7 @@ TEST_F(GenericEnd2endTest, SimpleBidiStreaming) { client_ok(10); EXPECT_EQ(send_response.message(), recv_response.message()); - EXPECT_TRUE(recv_status.IsOk()); + EXPECT_TRUE(recv_status.ok()); } } // namespace diff --git a/test/cpp/end2end/mock_test.cc b/test/cpp/end2end/mock_test.cc index 0226da672c..2809ab8d3c 100644 --- a/test/cpp/end2end/mock_test.cc +++ b/test/cpp/end2end/mock_test.cc @@ -168,7 +168,7 @@ class FakeClient { request.set_message("hello world"); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(request.message(), response.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } void DoBidiStream() { @@ -199,7 +199,7 @@ class FakeClient { EXPECT_FALSE(stream->Read(&response)); Status s = stream->Finish(); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } void ResetStub(TestService::StubInterface* stub) { stub_ = stub; } diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index 5ee29e40f4..0b43dfd106 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -99,13 +99,13 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { gpr_now(), gpr_time_from_micros(request->param().client_cancel_after_us()))); } - return Status::Cancelled; + return Status::CANCELLED; } else if (request->has_param() && request->param().server_cancel_after_us()) { gpr_sleep_until(gpr_time_add( gpr_now(), gpr_time_from_micros(request->param().server_cancel_after_us()))); - return Status::Cancelled; + return Status::CANCELLED; } else { EXPECT_FALSE(context->IsCancelled()); } @@ -219,7 +219,7 @@ static void SendRpc(grpc::cpp::test::util::TestService::Stub* stub, ClientContext context; Status s = stub->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); } } diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index e351059246..f08f90b139 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -65,11 +65,11 @@ InteropClient::InteropClient(std::shared_ptr<ChannelInterface> channel) : channel_(channel) {} void InteropClient::AssertOkOrPrintErrorStatus(const Status& s) { - if (s.IsOk()) { + if (s.ok()) { return; } - gpr_log(GPR_INFO, "Error status code: %d, message: %s", s.code(), - s.details().c_str()); + gpr_log(GPR_INFO, "Error status code: %d, message: %s", s.error_code(), + s.error_message().c_str()); GPR_ASSERT(0); } @@ -321,7 +321,7 @@ void InteropClient::DoCancelAfterBegin() { gpr_log(GPR_INFO, "Trying to cancel..."); context.TryCancel(); Status s = stream->Finish(); - GPR_ASSERT(s.code() == StatusCode::CANCELLED); + GPR_ASSERT(s.error_code() == StatusCode::CANCELLED); gpr_log(GPR_INFO, "Canceling streaming done."); } diff --git a/test/cpp/interop/interop_test.cc b/test/cpp/interop/interop_test.cc index a7a5cc0b2c..aac6e56b89 100644 --- a/test/cpp/interop/interop_test.cc +++ b/test/cpp/interop/interop_test.cc @@ -52,6 +52,7 @@ extern "C" { #include <grpc/support/alloc.h> #include <grpc/support/host_port.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> #include "test/core/util/port.h" int test_client(const char* root, const char* host, int port) { diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index dc3a9f2ac5..28cd32a197 100644 --- a/test/cpp/qps/client.h +++ b/test/cpp/qps/client.h @@ -35,6 +35,7 @@ #define TEST_QPS_CLIENT_H #include "test/cpp/qps/histogram.h" +#include "test/cpp/qps/interarrival.h" #include "test/cpp/qps/timer.h" #include "test/cpp/qps/qpstest.grpc.pb.h" @@ -42,11 +43,31 @@ #include <mutex> namespace grpc { + +#if defined(__APPLE__) +// Specialize Timepoint for high res clock as we need that +template <> +class TimePoint<std::chrono::high_resolution_clock::time_point> { + public: + TimePoint(const std::chrono::high_resolution_clock::time_point& time) { + TimepointHR2Timespec(time, &time_); + } + gpr_timespec raw_time() const { return time_; } + + private: + gpr_timespec time_; +}; +#endif + namespace testing { +typedef std::chrono::high_resolution_clock grpc_time_source; +typedef std::chrono::time_point<grpc_time_source> grpc_time; + class Client { public: - explicit Client(const ClientConfig& config) : timer_(new Timer) { + explicit Client(const ClientConfig& config) + : timer_(new Timer), interarrival_timer_() { for (int i = 0; i < config.client_channels(); i++) { channels_.push_back(ClientChannelInfo( config.server_targets(i % config.server_targets_size()), config)); @@ -81,6 +102,7 @@ class Client { protected: SimpleRequest request_; + bool closed_loop_; class ClientChannelInfo { public: @@ -106,6 +128,61 @@ class Client { virtual bool ThreadFunc(Histogram* histogram, size_t thread_idx) = 0; + void SetupLoadTest(const ClientConfig& config, size_t num_threads) { + // Set up the load distribution based on the number of threads + if (config.load_type() == CLOSED_LOOP) { + closed_loop_ = true; + } else { + closed_loop_ = false; + + std::unique_ptr<RandomDist> random_dist; + const auto& load = config.load_params(); + switch (config.load_type()) { + case POISSON: + random_dist.reset( + new ExpDist(load.poisson().offered_load() / num_threads)); + break; + case UNIFORM: + random_dist.reset( + new UniformDist(load.uniform().interarrival_lo() * num_threads, + load.uniform().interarrival_hi() * num_threads)); + break; + case DETERMINISTIC: + random_dist.reset( + new DetDist(num_threads / load.determ().offered_load())); + break; + case PARETO: + random_dist.reset( + new ParetoDist(load.pareto().interarrival_base() * num_threads, + load.pareto().alpha())); + break; + default: + GPR_ASSERT(false); + break; + } + + interarrival_timer_.init(*random_dist, num_threads); + for (size_t i = 0; i < num_threads; i++) { + next_time_.push_back( + grpc_time_source::now() + + std::chrono::duration_cast<grpc_time_source::duration>( + interarrival_timer_(i))); + } + } + } + + bool NextIssueTime(int thread_idx, grpc_time* time_delay) { + if (closed_loop_) { + return false; + } else { + *time_delay = next_time_[thread_idx]; + next_time_[thread_idx] += + std::chrono::duration_cast<grpc_time_source::duration>( + interarrival_timer_(thread_idx)); + return true; + } + } + private: class Thread { public: @@ -168,6 +245,9 @@ class Client { std::vector<std::unique_ptr<Thread>> threads_; std::unique_ptr<Timer> timer_; + + InterarrivalTimer interarrival_timer_; + std::vector<grpc_time> next_time_; }; std::unique_ptr<Client> CreateSynchronousUnaryClient(const ClientConfig& args); diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc index 00bbd8a8a0..1b7a8d26b2 100644 --- a/test/cpp/qps/client_async.cc +++ b/test/cpp/qps/client_async.cc @@ -32,8 +32,11 @@ */ #include <cassert> +#include <forward_list> #include <functional> +#include <list> #include <memory> +#include <mutex> #include <string> #include <thread> #include <vector> @@ -55,38 +58,55 @@ namespace grpc { namespace testing { +typedef std::list<grpc_time> deadline_list; + class ClientRpcContext { public: - ClientRpcContext() {} + explicit ClientRpcContext(int ch) : channel_id_(ch) {} virtual ~ClientRpcContext() {} // next state, return false if done. Collect stats when appropriate virtual bool RunNextState(bool, Histogram* hist) = 0; - virtual void StartNewClone() = 0; + virtual ClientRpcContext* StartNewClone() = 0; static void* tag(ClientRpcContext* c) { return reinterpret_cast<void*>(c); } static ClientRpcContext* detag(void* t) { return reinterpret_cast<ClientRpcContext*>(t); } + + deadline_list::iterator deadline_posn() const { return deadline_posn_; } + void set_deadline_posn(const deadline_list::iterator& it) { + deadline_posn_ = it; + } + virtual void Start(CompletionQueue* cq) = 0; + int channel_id() const { return channel_id_; } + + protected: + int channel_id_; + + private: + deadline_list::iterator deadline_posn_; }; template <class RequestType, class ResponseType> class ClientRpcContextUnaryImpl : public ClientRpcContext { public: ClientRpcContextUnaryImpl( - TestService::Stub* stub, const RequestType& req, + int channel_id, TestService::Stub* stub, const RequestType& req, std::function< std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( - TestService::Stub*, grpc::ClientContext*, const RequestType&)> - start_req, + TestService::Stub*, grpc::ClientContext*, const RequestType&, + CompletionQueue*)> start_req, std::function<void(grpc::Status, ResponseType*)> on_done) - : context_(), + : ClientRpcContext(channel_id), + context_(), stub_(stub), req_(req), response_(), next_state_(&ClientRpcContextUnaryImpl::RespDone), callback_(on_done), - start_req_(start_req), - start_(Timer::Now()), - response_reader_(start_req(stub_, &context_, req_)) { + start_req_(start_req) {} + void Start(CompletionQueue* cq) GRPC_OVERRIDE { + start_ = Timer::Now(); + response_reader_ = start_req_(stub_, &context_, req_, cq); response_reader_->Finish(&response_, &status_, ClientRpcContext::tag(this)); } ~ClientRpcContextUnaryImpl() GRPC_OVERRIDE {} @@ -98,8 +118,9 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { return ret; } - void StartNewClone() GRPC_OVERRIDE { - new ClientRpcContextUnaryImpl(stub_, req_, start_req_, callback_); + ClientRpcContext* StartNewClone() GRPC_OVERRIDE { + return new ClientRpcContextUnaryImpl(channel_id_, stub_, req_, start_req_, + callback_); } private: @@ -109,7 +130,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { } bool DoCallBack(bool) { callback_(status_, &response_); - return false; + return true; // we're done, this'll be ignored } grpc::ClientContext context_; TestService::Stub* stub_; @@ -118,29 +139,54 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { bool (ClientRpcContextUnaryImpl::*next_state_)(bool); std::function<void(grpc::Status, ResponseType*)> callback_; std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( - TestService::Stub*, grpc::ClientContext*, const RequestType&)> start_req_; + TestService::Stub*, grpc::ClientContext*, const RequestType&, + CompletionQueue*)> start_req_; grpc::Status status_; double start_; std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>> response_reader_; }; +typedef std::forward_list<ClientRpcContext*> context_list; + class AsyncClient : public Client { public: - explicit AsyncClient(const ClientConfig& config, - std::function<void(CompletionQueue*, TestService::Stub*, - const SimpleRequest&)> setup_ctx) - : Client(config) { + explicit AsyncClient( + const ClientConfig& config, + std::function<ClientRpcContext*(int, TestService::Stub*, + const SimpleRequest&)> setup_ctx) + : Client(config), + channel_lock_(config.client_channels()), + contexts_(config.client_channels()), + max_outstanding_per_channel_(config.outstanding_rpcs_per_channel()), + channel_count_(config.client_channels()), + pref_channel_inc_(config.async_client_threads()) { + SetupLoadTest(config, config.async_client_threads()); + for (int i = 0; i < config.async_client_threads(); i++) { cli_cqs_.emplace_back(new CompletionQueue); + if (!closed_loop_) { + rpc_deadlines_.emplace_back(); + next_channel_.push_back(i % channel_count_); + issue_allowed_.push_back(true); + + grpc_time next_issue; + NextIssueTime(i, &next_issue); + next_issue_.push_back(next_issue); + } } + int t = 0; for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) { - for (auto channel = channels_.begin(); channel != channels_.end(); - channel++) { + for (int ch = 0; ch < channel_count_; ch++) { auto* cq = cli_cqs_[t].get(); t = (t + 1) % cli_cqs_.size(); - setup_ctx(cq, channel->get_stub(), request_); + auto ctx = setup_ctx(ch, channels_[ch].get_stub(), request_); + if (closed_loop_) { + ctx->Start(cq); + } else { + contexts_[ch].push_front(ctx); + } } } } @@ -159,30 +205,126 @@ class AsyncClient : public Client { 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))) { + grpc_time deadline, short_deadline; + if (closed_loop_) { + deadline = grpc_time_source::now() + std::chrono::seconds(1); + short_deadline = deadline; + } else { + if (rpc_deadlines_[thread_idx].empty()) { + deadline = grpc_time_source::now() + std::chrono::seconds(1); + } else { + deadline = *(rpc_deadlines_[thread_idx].begin()); + } + short_deadline = + issue_allowed_[thread_idx] ? next_issue_[thread_idx] : deadline; + } + + bool got_event; + + switch (cli_cqs_[thread_idx]->AsyncNext(&got_tag, &ok, short_deadline)) { case CompletionQueue::SHUTDOWN: return false; case CompletionQueue::TIMEOUT: - return true; + got_event = false; + break; case CompletionQueue::GOT_EVENT: + got_event = true; + break; + default: + GPR_ASSERT(false); 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; + if ((closed_loop_ || !rpc_deadlines_[thread_idx].empty()) && + grpc_time_source::now() > deadline) { + // we have missed some 1-second deadline, which is worth noting + gpr_log(GPR_INFO, "Missed an RPC deadline"); + // Don't give up, as there might be some truly heavy tails + } + if (got_event) { + ClientRpcContext* ctx = ClientRpcContext::detag(got_tag); + if (ctx->RunNextState(ok, histogram) == false) { + // call the callback and then clone the ctx + ctx->RunNextState(ok, histogram); + ClientRpcContext* clone_ctx = ctx->StartNewClone(); + if (closed_loop_) { + clone_ctx->Start(cli_cqs_[thread_idx].get()); + } else { + // Remove the entry from the rpc deadlines list + rpc_deadlines_[thread_idx].erase(ctx->deadline_posn()); + // Put the clone_ctx in the list of idle contexts for this channel + // Under lock + int ch = clone_ctx->channel_id(); + std::lock_guard<std::mutex> g(channel_lock_[ch]); + contexts_[ch].push_front(clone_ctx); + } + // delete the old version + delete ctx; + } + if (!closed_loop_) + issue_allowed_[thread_idx] = + true; // may be ok now even if it hadn't been + } + if (!closed_loop_ && issue_allowed_[thread_idx] && + grpc_time_source::now() >= next_issue_[thread_idx]) { + // Attempt to issue + bool issued = false; + for (int num_attempts = 0, channel_attempt = next_channel_[thread_idx]; + num_attempts < channel_count_ && !issued; num_attempts++) { + bool can_issue = false; + ClientRpcContext* ctx = nullptr; + { + std::lock_guard<std::mutex> g(channel_lock_[channel_attempt]); + if (!contexts_[channel_attempt].empty()) { + // Get an idle context from the front of the list + ctx = *(contexts_[channel_attempt].begin()); + contexts_[channel_attempt].pop_front(); + can_issue = true; + } + } + if (can_issue) { + // do the work to issue + rpc_deadlines_[thread_idx].emplace_back(grpc_time_source::now() + + std::chrono::seconds(1)); + auto it = rpc_deadlines_[thread_idx].end(); + --it; + ctx->set_deadline_posn(it); + ctx->Start(cli_cqs_[thread_idx].get()); + issued = true; + // If we did issue, then next time, try our thread's next + // preferred channel + next_channel_[thread_idx] += pref_channel_inc_; + if (next_channel_[thread_idx] >= channel_count_) + next_channel_[thread_idx] = (thread_idx % channel_count_); + } else { + // Do a modular increment of channel attempt if we couldn't issue + channel_attempt = (channel_attempt + 1) % channel_count_; + } + } + if (issued) { + // We issued one; see when we can issue the next + grpc_time next_issue; + NextIssueTime(thread_idx, &next_issue); + next_issue_[thread_idx] = next_issue; + } else { + issue_allowed_[thread_idx] = false; + } } - return true; } private: std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_; + + std::vector<deadline_list> rpc_deadlines_; // per thread deadlines + std::vector<int> next_channel_; // per thread round-robin channel ctr + std::vector<bool> issue_allowed_; // may this thread attempt to issue + std::vector<grpc_time> next_issue_; // when should it issue? + + std::vector<std::mutex> channel_lock_; + std::vector<context_list> contexts_; // per-channel list of idle contexts + int max_outstanding_per_channel_; + int channel_count_; + int pref_channel_inc_; }; class AsyncUnaryClient GRPC_FINAL : public AsyncClient { @@ -194,15 +336,15 @@ class AsyncUnaryClient GRPC_FINAL : public AsyncClient { ~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); } private: - static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub, - const SimpleRequest& req) { + static ClientRpcContext* SetupCtx(int channel_id, 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) { + auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx, + const SimpleRequest& request, CompletionQueue* cq) { return stub->AsyncUnaryCall(ctx, request, cq); }; - new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( - stub, req, start_req, check_done); + return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( + channel_id, stub, req, start_req, check_done); } }; @@ -210,26 +352,30 @@ template <class RequestType, class ResponseType> class ClientRpcContextStreamingImpl : public ClientRpcContext { public: ClientRpcContextStreamingImpl( - TestService::Stub* stub, const RequestType& req, - std::function<std::unique_ptr< - grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( - TestService::Stub*, grpc::ClientContext*, void*)> start_req, + int channel_id, TestService::Stub* stub, const RequestType& req, + std::function<std::unique_ptr<grpc::ClientAsyncReaderWriter< + RequestType, ResponseType>>(TestService::Stub*, grpc::ClientContext*, + CompletionQueue*, void*)> start_req, std::function<void(grpc::Status, ResponseType*)> on_done) - : context_(), + : ClientRpcContext(channel_id), + context_(), stub_(stub), req_(req), response_(), next_state_(&ClientRpcContextStreamingImpl::ReqSent), callback_(on_done), start_req_(start_req), - start_(Timer::Now()), - stream_(start_req_(stub_, &context_, ClientRpcContext::tag(this))) {} + start_(Timer::Now()) {} ~ClientRpcContextStreamingImpl() GRPC_OVERRIDE {} bool RunNextState(bool ok, Histogram* hist) GRPC_OVERRIDE { return (this->*next_state_)(ok, hist); } - void StartNewClone() GRPC_OVERRIDE { - new ClientRpcContextStreamingImpl(stub_, req_, start_req_, callback_); + ClientRpcContext* StartNewClone() GRPC_OVERRIDE { + return new ClientRpcContextStreamingImpl(channel_id_, stub_, req_, + start_req_, callback_); + } + void Start(CompletionQueue* cq) GRPC_OVERRIDE { + stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this)); } private: @@ -263,7 +409,8 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext { std::function<void(grpc::Status, ResponseType*)> callback_; std::function< std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( - TestService::Stub*, grpc::ClientContext*, void*)> start_req_; + TestService::Stub*, grpc::ClientContext*, CompletionQueue*, void*)> + start_req_; grpc::Status status_; double start_; std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>> @@ -274,22 +421,25 @@ class AsyncStreamingClient GRPC_FINAL : public AsyncClient { public: explicit AsyncStreamingClient(const ClientConfig& config) : AsyncClient(config, SetupCtx) { + // async streaming currently only supported closed loop + GPR_ASSERT(config.load_type() == CLOSED_LOOP); + StartThreads(config.async_client_threads()); } ~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); } private: - static void SetupCtx(CompletionQueue* cq, TestService::Stub* stub, - const SimpleRequest& req) { + static ClientRpcContext* SetupCtx(int channel_id, 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 start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx, + CompletionQueue* cq, void* tag) { auto stream = stub->AsyncStreamingCall(ctx, cq, tag); return stream; }; - new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( - stub, req, start_req, check_done); + return new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( + channel_id, stub, req, start_req, check_done); } }; diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc index c28dc91321..718698bfe1 100644 --- a/test/cpp/qps/client_sync.cc +++ b/test/cpp/qps/client_sync.cc @@ -32,6 +32,7 @@ */ #include <cassert> +#include <chrono> #include <memory> #include <mutex> #include <string> @@ -57,6 +58,7 @@ #include "test/cpp/qps/client.h" #include "test/cpp/qps/qpstest.grpc.pb.h" #include "test/cpp/qps/histogram.h" +#include "test/cpp/qps/interarrival.h" #include "test/cpp/qps/timer.h" namespace grpc { @@ -68,11 +70,19 @@ class SynchronousClient : public Client { num_threads_ = config.outstanding_rpcs_per_channel() * config.client_channels(); responses_.resize(num_threads_); + SetupLoadTest(config, num_threads_); } virtual ~SynchronousClient(){}; protected: + void WaitToIssue(int thread_idx) { + grpc_time next_time; + if (NextIssueTime(thread_idx, &next_time)) { + std::this_thread::sleep_until(next_time); + } + } + size_t num_threads_; std::vector<SimpleResponse> responses_; }; @@ -86,13 +96,14 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient { ~SynchronousUnaryClient() { EndThreads(); } bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE { + WaitToIssue(thread_idx); auto* stub = channels_[thread_idx % channels_.size()].get_stub(); double start = Timer::Now(); grpc::ClientContext context; grpc::Status s = stub->UnaryCall(&context, request_, &responses_[thread_idx]); histogram->Add((Timer::Now() - start) * 1e9); - return s.IsOk(); + return s.ok(); } }; @@ -113,12 +124,13 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient { for (auto stream = stream_.begin(); stream != stream_.end(); stream++) { if (*stream) { (*stream)->WritesDone(); - EXPECT_TRUE((*stream)->Finish().IsOk()); + EXPECT_TRUE((*stream)->Finish().ok()); } } } bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE { + WaitToIssue(thread_idx); double start = Timer::Now(); if (stream_[thread_idx]->Write(request_) && stream_[thread_idx]->Read(&responses_[thread_idx])) { diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index bf12730f97..c8cc11e6ab 100644 --- a/test/cpp/qps/driver.cc +++ b/test/cpp/qps/driver.cc @@ -241,11 +241,11 @@ std::unique_ptr<ScenarioResult> RunScenario( for (auto client = clients.begin(); client != clients.end(); client++) { GPR_ASSERT(client->stream->WritesDone()); - GPR_ASSERT(client->stream->Finish().IsOk()); + GPR_ASSERT(client->stream->Finish().ok()); } for (auto server = servers.begin(); server != servers.end(); server++) { GPR_ASSERT(server->stream->WritesDone()); - GPR_ASSERT(server->stream->Finish().IsOk()); + GPR_ASSERT(server->stream->Finish().ok()); } return result; } diff --git a/test/cpp/qps/interarrival.h b/test/cpp/qps/interarrival.h new file mode 100644 index 0000000000..f90a17a894 --- /dev/null +++ b/test/cpp/qps/interarrival.h @@ -0,0 +1,178 @@ +/* + * + * 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. + * + */ + +#ifndef TEST_QPS_INTERARRIVAL_H +#define TEST_QPS_INTERARRIVAL_H + +#include <chrono> +#include <cmath> +#include <random> + +#include <grpc++/config.h> + +namespace grpc { +namespace testing { + +// First create classes that define a random distribution +// Note that this code does not include C++-specific random distribution +// features supported in std::random. Although this would make this code easier, +// this code is required to serve as the template code for other language +// stacks. Thus, this code only uses a uniform distribution of doubles [0,1) +// and then provides the distribution functions itself. + +class RandomDist { + public: + RandomDist() {} + virtual ~RandomDist() = 0; + // Argument to operator() is a uniform double in the range [0,1) + virtual double operator()(double uni) const = 0; +}; + +inline RandomDist::~RandomDist() {} + +// ExpDist implements an exponential distribution, which is the +// interarrival distribution for a Poisson process. The parameter +// lambda is the mean rate of arrivals. This is the +// most useful distribution since it is actually additive and +// memoryless. It is a good representation of activity coming in from +// independent identical stationary sources. For more information, +// see http://en.wikipedia.org/wiki/Exponential_distribution + +class ExpDist GRPC_FINAL : public RandomDist { + public: + explicit ExpDist(double lambda) : lambda_recip_(1.0 / lambda) {} + ~ExpDist() GRPC_OVERRIDE {} + double operator()(double uni) const GRPC_OVERRIDE { + // Note: Use 1.0-uni above to avoid NaN if uni is 0 + return lambda_recip_ * (-log(1.0 - uni)); + } + + private: + double lambda_recip_; +}; + +// UniformDist implements a random distribution that has +// interarrival time uniformly spread between [lo,hi). The +// mean interarrival time is (lo+hi)/2. For more information, +// see http://en.wikipedia.org/wiki/Uniform_distribution_%28continuous%29 + +class UniformDist GRPC_FINAL : public RandomDist { + public: + UniformDist(double lo, double hi) : lo_(lo), range_(hi - lo) {} + ~UniformDist() GRPC_OVERRIDE {} + double operator()(double uni) const GRPC_OVERRIDE { + return uni * range_ + lo_; + } + + private: + double lo_; + double range_; +}; + +// DetDist provides a random distribution with interarrival time +// of val. Note that this is not additive, so using this on multiple +// flows of control (threads within the same client or separate +// clients) will not preserve any deterministic interarrival gap across +// requests. + +class DetDist GRPC_FINAL : public RandomDist { + public: + explicit DetDist(double val) : val_(val) {} + ~DetDist() GRPC_OVERRIDE {} + double operator()(double uni) const GRPC_OVERRIDE { return val_; } + + private: + double val_; +}; + +// ParetoDist provides a random distribution with interarrival time +// spread according to a Pareto (heavy-tailed) distribution. In this +// model, many interarrival times are close to the base, but a sufficient +// number will be high (up to infinity) as to disturb the mean. It is a +// good representation of the response times of data center jobs. See +// http://en.wikipedia.org/wiki/Pareto_distribution + +class ParetoDist GRPC_FINAL : public RandomDist { + public: + ParetoDist(double base, double alpha) + : base_(base), alpha_recip_(1.0 / alpha) {} + ~ParetoDist() GRPC_OVERRIDE {} + double operator()(double uni) const GRPC_OVERRIDE { + // Note: Use 1.0-uni above to avoid div by zero if uni is 0 + return base_ / pow(1.0 - uni, alpha_recip_); + } + + private: + double base_; + double alpha_recip_; +}; + +// A class library for generating pseudo-random interarrival times +// in an efficient re-entrant way. The random table is built at construction +// time, and each call must include the thread id of the invoker + +typedef std::default_random_engine qps_random_engine; + +class InterarrivalTimer { + public: + InterarrivalTimer() {} + void init(const RandomDist& r, int threads, int entries = 1000000) { + qps_random_engine gen; + std::uniform_real_distribution<double> uniform(0.0, 1.0); + for (int i = 0; i < entries; i++) { + random_table_.push_back(std::chrono::nanoseconds( + static_cast<int64_t>(1e9 * r(uniform(gen))))); + } + // Now set up the thread positions + for (int i = 0; i < threads; i++) { + thread_posns_.push_back(random_table_.begin() + (entries * i) / threads); + } + } + virtual ~InterarrivalTimer(){}; + + std::chrono::nanoseconds operator()(int thread_num) { + auto ret = *(thread_posns_[thread_num]++); + if (thread_posns_[thread_num] == random_table_.end()) + thread_posns_[thread_num] = random_table_.begin(); + return ret; + } + + private: + typedef std::vector<std::chrono::nanoseconds> time_table; + std::vector<time_table::const_iterator> thread_posns_; + time_table random_table_; +}; +} +} + +#endif diff --git a/test/cpp/qps/qps_driver.cc b/test/cpp/qps/qps_driver.cc index 281e2e8119..d534846365 100644 --- a/test/cpp/qps/qps_driver.cc +++ b/test/cpp/qps/qps_driver.cc @@ -63,11 +63,15 @@ DEFINE_int32(client_channels, 1, "Number of client channels"); DEFINE_int32(payload_size, 1, "Payload size"); DEFINE_string(client_type, "SYNCHRONOUS_CLIENT", "Client type"); DEFINE_int32(async_client_threads, 1, "Async client threads"); +DEFINE_string(load_type, "CLOSED_LOOP", "Load type"); +DEFINE_double(load_param_1, 0.0, "Load parameter 1"); +DEFINE_double(load_param_2, 0.0, "Load parameter 2"); using grpc::testing::ClientConfig; using grpc::testing::ServerConfig; using grpc::testing::ClientType; using grpc::testing::ServerType; +using grpc::testing::LoadType; using grpc::testing::RpcType; using grpc::testing::ResourceUsage; @@ -80,11 +84,14 @@ static void QpsDriver() { ClientType client_type; ServerType server_type; + LoadType load_type; GPR_ASSERT(ClientType_Parse(FLAGS_client_type, &client_type)); GPR_ASSERT(ServerType_Parse(FLAGS_server_type, &server_type)); + GPR_ASSERT(LoadType_Parse(FLAGS_load_type, &load_type)); ClientConfig client_config; client_config.set_client_type(client_type); + client_config.set_load_type(load_type); client_config.set_enable_ssl(FLAGS_enable_ssl); client_config.set_outstanding_rpcs_per_channel( FLAGS_outstanding_rpcs_per_channel); @@ -93,6 +100,43 @@ static void QpsDriver() { client_config.set_async_client_threads(FLAGS_async_client_threads); client_config.set_rpc_type(rpc_type); + // set up the load parameters + switch (load_type) { + case grpc::testing::CLOSED_LOOP: + break; + case grpc::testing::POISSON: { + auto poisson = client_config.mutable_load_params()->mutable_poisson(); + GPR_ASSERT(FLAGS_load_param_1 != 0.0); + poisson->set_offered_load(FLAGS_load_param_1); + break; + } + case grpc::testing::UNIFORM: { + auto uniform = client_config.mutable_load_params()->mutable_uniform(); + GPR_ASSERT(FLAGS_load_param_1 != 0.0); + GPR_ASSERT(FLAGS_load_param_2 != 0.0); + uniform->set_interarrival_lo(FLAGS_load_param_1 / 1e6); + uniform->set_interarrival_hi(FLAGS_load_param_2 / 1e6); + break; + } + case grpc::testing::DETERMINISTIC: { + auto determ = client_config.mutable_load_params()->mutable_determ(); + GPR_ASSERT(FLAGS_load_param_1 != 0.0); + determ->set_offered_load(FLAGS_load_param_1); + break; + } + case grpc::testing::PARETO: { + auto pareto = client_config.mutable_load_params()->mutable_pareto(); + GPR_ASSERT(FLAGS_load_param_1 != 0.0); + GPR_ASSERT(FLAGS_load_param_2 != 0.0); + pareto->set_interarrival_base(FLAGS_load_param_1 / 1e6); + pareto->set_alpha(FLAGS_load_param_2); + break; + } + default: + GPR_ASSERT(false); + break; + } + ServerConfig server_config; server_config.set_server_type(server_type); server_config.set_threads(FLAGS_server_threads); @@ -112,7 +156,7 @@ static void QpsDriver() { FLAGS_warmup_seconds, FLAGS_benchmark_seconds, FLAGS_local_workers); GetReporter()->ReportQPS(*result); - GetReporter()->ReportQPSPerCore(*result, server_config); + GetReporter()->ReportQPSPerCore(*result); GetReporter()->ReportLatency(*result); GetReporter()->ReportTimes(*result); } diff --git a/test/cpp/qps/qps_interarrival_test.cc b/test/cpp/qps/qps_interarrival_test.cc new file mode 100644 index 0000000000..cecd1be03f --- /dev/null +++ b/test/cpp/qps/qps_interarrival_test.cc @@ -0,0 +1,76 @@ +/* + * + * 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/cpp/qps/interarrival.h" +#include <chrono> +#include <iostream> + +// Use the C histogram rather than C++ to avoid depending on proto +#include <grpc/support/histogram.h> +#include <grpc++/config.h> + +using grpc::testing::RandomDist; +using grpc::testing::InterarrivalTimer; + +void RunTest(RandomDist&& r, int threads, std::string title) { + InterarrivalTimer timer; + timer.init(r, threads); + gpr_histogram *h(gpr_histogram_create(0.01, 60e9)); + + for (int i = 0; i < 10000000; i++) { + for (int j = 0; j < threads; j++) { + gpr_histogram_add(h, timer(j).count()); + } + } + + std::cout << title << " Distribution" << std::endl; + std::cout << "Value, Percentile" << std::endl; + for (double pct = 0.0; pct < 100.0; pct += 1.0) { + std::cout << gpr_histogram_percentile(h, pct) << "," << pct << std::endl; + } + + gpr_histogram_destroy(h); +} + +using grpc::testing::ExpDist; +using grpc::testing::DetDist; +using grpc::testing::UniformDist; +using grpc::testing::ParetoDist; + +int main(int argc, char **argv) { + RunTest(ExpDist(10.0), 5, std::string("Exponential(10)")); + RunTest(DetDist(5.0), 5, std::string("Det(5)")); + RunTest(UniformDist(0.0, 10.0), 5, std::string("Uniform(1,10)")); + RunTest(ParetoDist(1.0, 1.0), 5, std::string("Pareto(1,1)")); + return 0; +} diff --git a/test/cpp/qps/qps_test.cc b/test/cpp/qps/qps_test.cc index 63a37ae07e..07b4834cc0 100644 --- a/test/cpp/qps/qps_test.cc +++ b/test/cpp/qps/qps_test.cc @@ -67,7 +67,7 @@ static void RunQPS() { const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); - GetReporter()->ReportQPSPerCore(*result, server_config); + GetReporter()->ReportQPSPerCore(*result); GetReporter()->ReportLatency(*result); } diff --git a/test/cpp/qps/qps_test_openloop.cc b/test/cpp/qps/qps_test_openloop.cc new file mode 100644 index 0000000000..52873b2987 --- /dev/null +++ b/test/cpp/qps/qps_test_openloop.cc @@ -0,0 +1,87 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <set> + +#include <grpc/support/log.h> + +#include <signal.h> + +#include "test/cpp/qps/driver.h" +#include "test/cpp/qps/report.h" +#include "test/cpp/util/benchmark_config.h" + +namespace grpc { +namespace testing { + +static const int WARMUP = 5; +static const int BENCHMARK = 10; + +static void RunQPS() { + gpr_log(GPR_INFO, "Running QPS test, open-loop"); + + ClientConfig client_config; + client_config.set_client_type(ASYNC_CLIENT); + client_config.set_enable_ssl(false); + client_config.set_outstanding_rpcs_per_channel(1000); + client_config.set_client_channels(8); + client_config.set_payload_size(1); + client_config.set_async_client_threads(8); + client_config.set_rpc_type(UNARY); + client_config.set_load_type(POISSON); + client_config.mutable_load_params()-> + mutable_poisson()->set_offered_load(10000.0); + + ServerConfig server_config; + server_config.set_server_type(ASYNC_SERVER); + server_config.set_enable_ssl(false); + server_config.set_threads(4); + + const auto result = + RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); + + GetReporter()->ReportQPSPerCore(*result); + GetReporter()->ReportLatency(*result); +} + +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc::testing::InitBenchmark(&argc, &argv, true); + + signal(SIGPIPE, SIG_IGN); + grpc::testing::RunQPS(); + + return 0; +} diff --git a/test/cpp/qps/qps_test_with_poll.cc b/test/cpp/qps/qps_test_with_poll.cc new file mode 100644 index 0000000000..90a8da8d11 --- /dev/null +++ b/test/cpp/qps/qps_test_with_poll.cc @@ -0,0 +1,90 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <set> + +#include <grpc/support/log.h> + +#include <signal.h> + +#include "test/cpp/qps/driver.h" +#include "test/cpp/qps/report.h" +#include "test/cpp/util/benchmark_config.h" + +extern "C" { +#include "src/core/iomgr/pollset_posix.h" +} + +namespace grpc { +namespace testing { + +static const int WARMUP = 5; +static const int BENCHMARK = 5; + +static void RunQPS() { + gpr_log(GPR_INFO, "Running QPS test"); + + ClientConfig client_config; + client_config.set_client_type(ASYNC_CLIENT); + client_config.set_enable_ssl(false); + client_config.set_outstanding_rpcs_per_channel(1000); + client_config.set_client_channels(8); + client_config.set_payload_size(1); + client_config.set_async_client_threads(8); + client_config.set_rpc_type(UNARY); + + ServerConfig server_config; + server_config.set_server_type(ASYNC_SERVER); + server_config.set_enable_ssl(false); + server_config.set_threads(4); + + const auto result = + RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); + + GetReporter()->ReportQPSPerCore(*result); + GetReporter()->ReportLatency(*result); +} + +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc::testing::InitBenchmark(&argc, &argv, true); + + grpc_platform_become_multipoller = grpc_poll_become_multipoller; + + signal(SIGPIPE, SIG_IGN); + grpc::testing::RunQPS(); + + return 0; +} diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc index fb49271991..423275ee85 100644 --- a/test/cpp/qps/qps_worker.cc +++ b/test/cpp/qps/qps_worker.cc @@ -71,6 +71,8 @@ std::unique_ptr<Client> CreateClient(const ClientConfig& config) { return (config.rpc_type() == RpcType::UNARY) ? CreateAsyncUnaryClient(config) : CreateAsyncStreamingClient(config); + default: + abort(); } abort(); } @@ -82,6 +84,8 @@ std::unique_ptr<Server> CreateServer(const ServerConfig& config, return CreateSynchronousServer(config, server_port); case ServerType::ASYNC_SERVER: return CreateAsyncServer(config, server_port); + default: + abort(); } abort(); } @@ -96,7 +100,7 @@ class WorkerImpl GRPC_FINAL : public Worker::Service { GRPC_OVERRIDE { InstanceGuard g(this); if (!g.Acquired()) { - return Status(RESOURCE_EXHAUSTED); + return Status(StatusCode::RESOURCE_EXHAUSTED, ""); } grpc_profiler_start("qps_client.prof"); @@ -110,7 +114,7 @@ class WorkerImpl GRPC_FINAL : public Worker::Service { GRPC_OVERRIDE { InstanceGuard g(this); if (!g.Acquired()) { - return Status(RESOURCE_EXHAUSTED); + return Status(StatusCode::RESOURCE_EXHAUSTED, ""); } grpc_profiler_start("qps_server.prof"); @@ -155,22 +159,22 @@ class WorkerImpl GRPC_FINAL : public Worker::Service { ServerReaderWriter<ClientStatus, ClientArgs>* stream) { ClientArgs args; if (!stream->Read(&args)) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } if (!args.has_setup()) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } auto client = CreateClient(args.setup()); if (!client) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } ClientStatus status; if (!stream->Write(status)) { - return Status(UNKNOWN); + return Status(StatusCode::UNKNOWN, ""); } while (stream->Read(&args)) { if (!args.has_mark()) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } *status.mutable_stats() = client->Mark(); stream->Write(status); @@ -183,23 +187,23 @@ class WorkerImpl GRPC_FINAL : public Worker::Service { ServerReaderWriter<ServerStatus, ServerArgs>* stream) { ServerArgs args; if (!stream->Read(&args)) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } if (!args.has_setup()) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } auto server = CreateServer(args.setup(), server_port_); if (!server) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } ServerStatus status; status.set_port(server_port_); if (!stream->Write(status)) { - return Status(UNKNOWN); + return Status(StatusCode::UNKNOWN, ""); } while (stream->Read(&args)) { if (!args.has_mark()) { - return Status(INVALID_ARGUMENT); + return Status(StatusCode::INVALID_ARGUMENT, ""); } *status.mutable_stats() = server->Mark(); stream->Write(status); diff --git a/test/cpp/qps/qpstest.proto b/test/cpp/qps/qpstest.proto index 122a7df1ac..ef1f9451e9 100644 --- a/test/cpp/qps/qpstest.proto +++ b/test/cpp/qps/qpstest.proto @@ -30,83 +30,121 @@ // An integration test service that covers all the method signature permutations // of unary/streaming requests/responses. -syntax = "proto2"; +syntax = "proto3"; package grpc.testing; enum PayloadType { // Compressable text format. - COMPRESSABLE= 1; + COMPRESSABLE = 0; // Uncompressable binary format. - UNCOMPRESSABLE = 2; + UNCOMPRESSABLE = 1; // Randomly chosen from all other formats defined in this enum. - RANDOM = 3; + RANDOM = 2; } message StatsRequest { // run number - optional int32 test_num = 1; + int32 test_num = 1; } message ServerStats { // wall clock time - required double time_elapsed = 1; + double time_elapsed = 1; // user time used by the server process and threads - required double time_user = 2; + double time_user = 2; // server time used by the server process and all threads - required double time_system = 3; + double time_system = 3; } message Payload { // The type of data in body. - optional PayloadType type = 1; + PayloadType type = 1; // Primary contents of payload. - optional bytes body = 2; + bytes body = 2; } message HistogramData { repeated uint32 bucket = 1; - required double min_seen = 2; - required double max_seen = 3; - required double sum = 4; - required double sum_of_squares = 5; - required double count = 6; + double min_seen = 2; + double max_seen = 3; + double sum = 4; + double sum_of_squares = 5; + double count = 6; } enum ClientType { - SYNCHRONOUS_CLIENT = 1; - ASYNC_CLIENT = 2; + SYNCHRONOUS_CLIENT = 0; + ASYNC_CLIENT = 1; } enum ServerType { - SYNCHRONOUS_SERVER = 1; - ASYNC_SERVER = 2; + SYNCHRONOUS_SERVER = 0; + ASYNC_SERVER = 1; } enum RpcType { - UNARY = 1; - STREAMING = 2; + UNARY = 0; + STREAMING = 1; +} + +enum LoadType { + CLOSED_LOOP = 0; + POISSON = 1; + UNIFORM = 2; + DETERMINISTIC = 3; + PARETO = 4; +} + +message PoissonParams { + double offered_load = 1; +} + +message UniformParams { + double interarrival_lo = 1; + double interarrival_hi = 2; +} + +message DeterministicParams { + double offered_load = 1; +} + +message ParetoParams { + double interarrival_base = 1; + double alpha = 2; +} + +message LoadParams { + oneof load { + PoissonParams poisson = 1; + UniformParams uniform = 2; + DeterministicParams determ = 3; + ParetoParams pareto = 4; + }; } message ClientConfig { repeated string server_targets = 1; - required ClientType client_type = 2; - optional bool enable_ssl = 3 [default=false]; - required int32 outstanding_rpcs_per_channel = 4; - required int32 client_channels = 5; - required int32 payload_size = 6; + ClientType client_type = 2; + bool enable_ssl = 3; + int32 outstanding_rpcs_per_channel = 4; + int32 client_channels = 5; + int32 payload_size = 6; // only for async client: - optional int32 async_client_threads = 7; - optional RpcType rpc_type = 8 [default=UNARY]; - optional string host = 9; + int32 async_client_threads = 7; + RpcType rpc_type = 8; + string host = 9; + LoadType load_type = 10; + LoadParams load_params = 11; } // Request current stats -message Mark {} +message Mark { +} message ClientArgs { oneof argtype { @@ -116,21 +154,21 @@ message ClientArgs { } message ClientStats { - required HistogramData latencies = 1; - required double time_elapsed = 3; - required double time_user = 4; - required double time_system = 5; + HistogramData latencies = 1; + double time_elapsed = 2; + double time_user = 3; + double time_system = 4; } message ClientStatus { - optional ClientStats stats = 1; + ClientStats stats = 1; } message ServerConfig { - required ServerType server_type = 1; - optional int32 threads = 2 [default=1]; - optional bool enable_ssl = 3 [default=false]; - optional string host = 4; + ServerType server_type = 1; + int32 threads = 2; + bool enable_ssl = 3; + string host = 4; } message ServerArgs { @@ -141,25 +179,25 @@ message ServerArgs { } message ServerStatus { - optional ServerStats stats = 1; - required int32 port = 2; + ServerStats stats = 1; + int32 port = 2; } message SimpleRequest { // Desired payload type in the response from the server. // If response_type is RANDOM, server randomly chooses one from other formats. - optional PayloadType response_type = 1 [default=COMPRESSABLE]; + PayloadType response_type = 1; // Desired payload size in the response from the server. // If response_type is COMPRESSABLE, this denotes the size before compression. - optional int32 response_size = 2 [default=0]; + int32 response_size = 2; // Optional input payload sent along with the request. - optional Payload payload = 3; + Payload payload = 3; } message SimpleResponse { - optional Payload payload = 1; + Payload payload = 1; } service TestService { diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc index e116175e3b..678ea080d1 100644 --- a/test/cpp/qps/report.cc +++ b/test/cpp/qps/report.cc @@ -49,10 +49,9 @@ void CompositeReporter::ReportQPS(const ScenarioResult& result) const { } } -void CompositeReporter::ReportQPSPerCore(const ScenarioResult& result, - const ServerConfig& config) const { +void CompositeReporter::ReportQPSPerCore(const ScenarioResult& result) const { for (size_t i = 0; i < reporters_.size(); ++i) { - reporters_[i]->ReportQPSPerCore(result, config); + reporters_[i]->ReportQPSPerCore(result); } } @@ -76,15 +75,14 @@ void GprLogReporter::ReportQPS(const ScenarioResult& result) const { [](ResourceUsage u) { return u.wall_time; })); } -void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result, - const ServerConfig& server_config) const { +void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result) const { auto qps = result.latencies.Count() / average(result.client_resources, [](ResourceUsage u) { return u.wall_time; }); gpr_log(GPR_INFO, "QPS: %.1f (%.1f/server core)", qps, - qps / server_config.threads()); + qps / result.server_config.threads()); } void GprLogReporter::ReportLatency(const ScenarioResult& result) const { diff --git a/test/cpp/qps/report.h b/test/cpp/qps/report.h index 630275ecda..0cce08816a 100644 --- a/test/cpp/qps/report.h +++ b/test/cpp/qps/report.h @@ -62,8 +62,7 @@ class Reporter { virtual void ReportQPS(const ScenarioResult& result) const = 0; /** Reports QPS per core as (YYY/server core). */ - virtual void ReportQPSPerCore(const ScenarioResult& result, - const ServerConfig& config) const = 0; + virtual void ReportQPSPerCore(const ScenarioResult& result) const = 0; /** Reports latencies for the 50, 90, 95, 99 and 99.9 percentiles, in ms. */ virtual void ReportLatency(const ScenarioResult& result) const = 0; @@ -84,8 +83,7 @@ class CompositeReporter : public Reporter { void add(std::unique_ptr<Reporter> reporter); void ReportQPS(const ScenarioResult& result) const GRPC_OVERRIDE; - void ReportQPSPerCore(const ScenarioResult& result, - const ServerConfig& config) const GRPC_OVERRIDE; + void ReportQPSPerCore(const ScenarioResult& result) const GRPC_OVERRIDE; void ReportLatency(const ScenarioResult& result) const GRPC_OVERRIDE; void ReportTimes(const ScenarioResult& result) const GRPC_OVERRIDE; @@ -100,8 +98,7 @@ class GprLogReporter : public Reporter { private: void ReportQPS(const ScenarioResult& result) const GRPC_OVERRIDE; - void ReportQPSPerCore(const ScenarioResult& result, - const ServerConfig& config) const GRPC_OVERRIDE; + void ReportQPSPerCore(const ScenarioResult& result) const GRPC_OVERRIDE; void ReportLatency(const ScenarioResult& result) const GRPC_OVERRIDE; void ReportTimes(const ScenarioResult& result) const GRPC_OVERRIDE; }; diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc index 977dfc2372..210aef4fd6 100644 --- a/test/cpp/qps/server_async.cc +++ b/test/cpp/qps/server_async.cc @@ -73,38 +73,43 @@ class AsyncQpsServerTest : public Server { gpr_free(server_address); builder.RegisterAsyncService(&async_service_); - srv_cq_ = builder.AddCompletionQueue(); + for (int i = 0; i < config.threads(); i++) { + srv_cqs_.emplace_back(std::move(builder.AddCompletionQueue())); + } server_ = builder.BuildAndStart(); using namespace std::placeholders; - request_unary_ = - std::bind(&TestService::AsyncService::RequestUnaryCall, &async_service_, - _1, _2, _3, srv_cq_.get(), srv_cq_.get(), _4); - request_streaming_ = - std::bind(&TestService::AsyncService::RequestStreamingCall, - &async_service_, _1, _2, srv_cq_.get(), srv_cq_.get(), _3); - for (int i = 0; i < 100; i++) { - contexts_.push_front( - new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( - request_unary_, ProcessRPC)); - contexts_.push_front( - new ServerRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( - request_streaming_, ProcessRPC)); + for (int i = 0; i < 10; i++) { + for (int j = 0; j < config.threads(); j++) { + auto request_unary = std::bind( + &TestService::AsyncService::RequestUnaryCall, &async_service_, _1, + _2, _3, srv_cqs_[j].get(), srv_cqs_[j].get(), _4); + auto request_streaming = std::bind( + &TestService::AsyncService::RequestStreamingCall, &async_service_, + _1, _2, srv_cqs_[j].get(), srv_cqs_[j].get(), _3); + contexts_.push_front( + new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>( + request_unary, ProcessRPC)); + contexts_.push_front( + new ServerRpcContextStreamingImpl<SimpleRequest, SimpleResponse>( + request_streaming, ProcessRPC)); + } } for (int i = 0; i < config.threads(); i++) { threads_.push_back(std::thread([=]() { // Wait until work is available or we are shutting down bool ok; void *got_tag; - while (srv_cq_->Next(&got_tag, &ok)) { + while (srv_cqs_[i]->Next(&got_tag, &ok)) { ServerRpcContext *ctx = detag(got_tag); // The tag is a pointer to an RPC context to invoke bool still_going = ctx->RunNextState(ok); - std::lock_guard<std::mutex> g(shutdown_mutex_); + std::unique_lock<std::mutex> g(shutdown_mutex_); if (!shutdown_) { // this RPC context is done, so refresh it if (!still_going) { + g.unlock(); ctx->Reset(); } } else { @@ -124,11 +129,13 @@ class AsyncQpsServerTest : public Server { for (auto thr = threads_.begin(); thr != threads_.end(); thr++) { thr->join(); } - srv_cq_->Shutdown(); - bool ok; - void *got_tag; - while (srv_cq_->Next(&got_tag, &ok)) - ; + for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) { + (*cq)->Shutdown(); + bool ok; + void *got_tag; + while ((*cq)->Next(&got_tag, &ok)) + ; + } while (!contexts_.empty()) { delete contexts_.front(); contexts_.pop_front(); @@ -305,15 +312,8 @@ class AsyncQpsServerTest : public Server { } std::vector<std::thread> threads_; std::unique_ptr<grpc::Server> server_; - std::unique_ptr<grpc::ServerCompletionQueue> srv_cq_; + std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> srv_cqs_; TestService::AsyncService async_service_; - std::function<void(ServerContext *, SimpleRequest *, - grpc::ServerAsyncResponseWriter<SimpleResponse> *, void *)> - request_unary_; - std::function<void( - ServerContext *, - grpc::ServerAsyncReaderWriter<SimpleResponse, SimpleRequest> *, void *)> - request_streaming_; std::forward_list<ServerRpcContext *> contexts_; std::mutex shutdown_mutex_; diff --git a/test/cpp/util/cli_call.cc b/test/cpp/util/cli_call.cc index eb67b8d314..83a7a1744a 100644 --- a/test/cpp/util/cli_call.cc +++ b/test/cpp/util/cli_call.cc @@ -52,11 +52,20 @@ namespace { void* tag(int i) { return (void*)(gpr_intptr) i; } } // namespace -void CliCall::Call(std::shared_ptr<grpc::ChannelInterface> channel, - const grpc::string& method, const grpc::string& request, - grpc::string* response) { +Status CliCall::Call(std::shared_ptr<grpc::ChannelInterface> channel, + const grpc::string& method, const grpc::string& request, + grpc::string* response, const MetadataContainer& metadata, + MetadataContainer* server_initial_metadata, + MetadataContainer* server_trailing_metadata) { std::unique_ptr<grpc::GenericStub> stub(new grpc::GenericStub(channel)); grpc::ClientContext ctx; + if (!metadata.empty()) { + for (std::multimap<grpc::string, grpc::string>::const_iterator iter = + metadata.begin(); + iter != metadata.end(); ++iter) { + ctx.AddMetadata(iter->first, iter->second); + } + } grpc::CompletionQueue cq; std::unique_ptr<grpc::GenericClientAsyncReaderWriter> call( stub->Call(&ctx, method, &cq, tag(1))); @@ -79,15 +88,14 @@ void CliCall::Call(std::shared_ptr<grpc::ChannelInterface> channel, cq.Next(&got_tag, &ok); if (!ok) { std::cout << "Failed to read response." << std::endl; - return; + return Status(StatusCode::INTERNAL, "Failed to read response"); } grpc::Status status; call->Finish(&status, tag(5)); cq.Next(&got_tag, &ok); GPR_ASSERT(ok); - if (status.IsOk()) { - std::cout << "RPC finished with OK status." << std::endl; + if (status.ok()) { std::vector<grpc::Slice> slices; recv_buffer.Dump(&slices); @@ -96,10 +104,10 @@ void CliCall::Call(std::shared_ptr<grpc::ChannelInterface> channel, response->append(reinterpret_cast<const char*>(slices[i].begin()), slices[i].size()); } - } else { - std::cout << "RPC finished with status code " << status.code() - << " details: " << status.details() << std::endl; } + *server_initial_metadata = ctx.GetServerInitialMetadata(); + *server_trailing_metadata = ctx.GetServerTrailingMetadata(); + return status; } } // namespace testing diff --git a/test/cpp/util/cli_call.h b/test/cpp/util/cli_call.h index 7be8bb63c4..8d114c9cb5 100644 --- a/test/cpp/util/cli_call.h +++ b/test/cpp/util/cli_call.h @@ -34,17 +34,23 @@ #ifndef GRPC_TEST_CPP_UTIL_CLI_CALL_H #define GRPC_TEST_CPP_UTIL_CLI_CALL_H +#include <map> + #include <grpc++/channel_interface.h> #include <grpc++/config.h> +#include <grpc++/status.h> namespace grpc { namespace testing { class CliCall GRPC_FINAL { public: - static void Call(std::shared_ptr<grpc::ChannelInterface> channel, - const grpc::string& method, const grpc::string& request, - grpc::string* response); + typedef std::multimap<grpc::string, grpc::string> MetadataContainer; + static Status Call(std::shared_ptr<grpc::ChannelInterface> channel, + const grpc::string& method, const grpc::string& request, + grpc::string* response, const MetadataContainer& metadata, + MetadataContainer* server_initial_metadata, + MetadataContainer* server_trailing_metadata); }; } // namespace testing diff --git a/test/cpp/util/cli_call_test.cc b/test/cpp/util/cli_call_test.cc index 457a5e77de..6cf86ea89b 100644 --- a/test/cpp/util/cli_call_test.cc +++ b/test/cpp/util/cli_call_test.cc @@ -60,6 +60,14 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { public: Status Echo(ServerContext* context, const EchoRequest* request, EchoResponse* response) GRPC_OVERRIDE { + if (!context->client_metadata().empty()) { + for (std::multimap<grpc::string, grpc::string>::const_iterator iter = + context->client_metadata().begin(); + iter != context->client_metadata().end(); ++iter) { + context->AddInitialMetadata(iter->first, iter->second); + } + } + context->AddTrailingMetadata("trailing_key", "trailing_value"); response->set_message(request->message()); return Status::OK; } @@ -106,16 +114,26 @@ TEST_F(CliCallTest, SimpleRpc) { request.set_message("Hello"); ClientContext context; + context.AddMetadata("key1", "val1"); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.IsOk()); + EXPECT_TRUE(s.ok()); const grpc::string kMethod("/grpc.cpp.test.util.TestService/Echo"); grpc::string request_bin, response_bin, expected_response_bin; EXPECT_TRUE(request.SerializeToString(&request_bin)); EXPECT_TRUE(response.SerializeToString(&expected_response_bin)); - CliCall::Call(channel_, kMethod, request_bin, &response_bin); + std::multimap<grpc::string, grpc::string> client_metadata, + server_initial_metadata, server_trailing_metadata; + client_metadata.insert(std::pair<grpc::string, grpc::string>("key1", "val1")); + Status s2 = CliCall::Call(channel_, kMethod, request_bin, &response_bin, + client_metadata, &server_initial_metadata, + &server_trailing_metadata); + EXPECT_TRUE(s2.ok()); + EXPECT_EQ(expected_response_bin, response_bin); + EXPECT_EQ(context.GetServerInitialMetadata(), server_initial_metadata); + EXPECT_EQ(context.GetServerTrailingMetadata(), server_trailing_metadata); } } // namespace testing diff --git a/test/cpp/util/grpc_cli.cc b/test/cpp/util/grpc_cli.cc index ad3c0af877..32d61b0307 100644 --- a/test/cpp/util/grpc_cli.cc +++ b/test/cpp/util/grpc_cli.cc @@ -41,8 +41,8 @@ body: "hello world" } b. under grpc/ run - protoc --proto_path=test/cpp/interop/ \ - --encode=grpc.testing.SimpleRequest test/cpp/interop/messages.proto \ + protoc --proto_path=test/proto/ \ + --encode=grpc.testing.SimpleRequest test/proto/messages.proto \ < input.txt > input.bin 2. Start a server make interop_server && bins/opt/interop_server --port=50051 @@ -51,10 +51,12 @@ /grpc.testing.TestService/UnaryCall --enable_ssl=false \ --input_binary_file=input.bin --output_binary_file=output.bin 4. Decode response - protoc --proto_path=test/cpp/interop/ \ - --decode=grpc.testing.SimpleResponse test/cpp/interop/messages.proto \ + protoc --proto_path=test/proto/ \ + --decode=grpc.testing.SimpleResponse test/proto/messages.proto \ < output.bin > output.txt 5. Now the text form of response should be in output.txt + Optionally, metadata can be passed to server via flag --metadata, e.g. + --metadata="MyHeaderKey1:Value1:MyHeaderKey2:Value2" */ #include <fstream> @@ -77,6 +79,44 @@ DEFINE_string(input_binary_file, "", "Path to input file containing serialized request."); DEFINE_string(output_binary_file, "output.bin", "Path to output file to write serialized response."); +DEFINE_string(metadata, "", + "Metadata to send to server, in the form of key1:val1:key2:val2"); + +void ParseMetadataFlag( + std::multimap<grpc::string, grpc::string>* client_metadata) { + if (FLAGS_metadata.empty()) { + return; + } + std::vector<grpc::string> fields; + grpc::string delim(":"); + size_t cur, next = -1; + do { + cur = next + 1; + next = FLAGS_metadata.find_first_of(delim, cur); + fields.push_back(FLAGS_metadata.substr(cur, next - cur)); + } while (next != grpc::string::npos); + if (fields.size() % 2) { + std::cout << "Failed to parse metadata flag" << std::endl; + exit(1); + } + for (size_t i = 0; i < fields.size(); i += 2) { + client_metadata->insert( + std::pair<grpc::string, grpc::string>(fields[i], fields[i + 1])); + } +} + +void PrintMetadata(const std::multimap<grpc::string, grpc::string>& m, + const grpc::string& message) { + if (m.empty()) { + return; + } + std::cout << message << std::endl; + for (std::multimap<grpc::string, grpc::string>::const_iterator iter = + m.begin(); + iter != m.end(); ++iter) { + std::cout << iter->first << " : " << iter->second << std::endl; + } +} int main(int argc, char** argv) { grpc::testing::InitTest(&argc, &argv, true); @@ -118,11 +158,27 @@ int main(int argc, char** argv) { grpc::CreateChannel(server_address, creds, grpc::ChannelArguments()); grpc::string response; - grpc::testing::CliCall::Call(channel, method, input_stream.str(), &response); - if (!response.empty()) { - std::ofstream output_file(FLAGS_output_binary_file, - std::ios::trunc | std::ios::binary); - output_file << response; + std::multimap<grpc::string, grpc::string> client_metadata, + server_initial_metadata, server_trailing_metadata; + ParseMetadataFlag(&client_metadata); + PrintMetadata(client_metadata, "Sending client initial metadata:"); + grpc::Status s = grpc::testing::CliCall::Call( + channel, method, input_stream.str(), &response, client_metadata, + &server_initial_metadata, &server_trailing_metadata); + PrintMetadata(server_initial_metadata, + "Received initial metadata from server:"); + PrintMetadata(server_trailing_metadata, + "Received trailing metadata from server:"); + if (s.ok()) { + std::cout << "Rpc succeeded with OK status" << std::endl; + if (!response.empty()) { + std::ofstream output_file(FLAGS_output_binary_file, + std::ios::trunc | std::ios::binary); + output_file << response; + } + } else { + std::cout << "Rpc failed with status code " << s.error_code() + << " error message " << s.error_message() << std::endl; } return 0; |