aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp')
-rw-r--r--test/cpp/microbenchmarks/bm_call_create.cc3
-rw-r--r--test/cpp/microbenchmarks/bm_cq_multiple_threads.cc4
-rw-r--r--test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc11
-rw-r--r--test/cpp/microbenchmarks/bm_fullstack_trickle.cc125
-rw-r--r--test/cpp/microbenchmarks/bm_pollset.cc36
-rw-r--r--test/cpp/performance/writes_per_rpc_test.cc4
-rw-r--r--test/cpp/qps/client_async.cc48
7 files changed, 180 insertions, 51 deletions
diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc
index c91219e98c..67e7c02535 100644
--- a/test/cpp/microbenchmarks/bm_call_create.cc
+++ b/test/cpp/microbenchmarks/bm_call_create.cc
@@ -563,7 +563,8 @@ static void BM_IsolatedFilter(benchmark::State &state) {
}
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- size_t channel_size = grpc_channel_stack_size(&filters[0], filters.size());
+ size_t channel_size = grpc_channel_stack_size(
+ filters.size() == 0 ? NULL : &filters[0], filters.size());
grpc_channel_stack *channel_stack =
static_cast<grpc_channel_stack *>(gpr_zalloc(channel_size));
GPR_ASSERT(GRPC_LOG_IF_ERROR(
diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
index 9d7f65d292..0d267da723 100644
--- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
+++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
@@ -67,7 +67,9 @@ static void pollset_init(grpc_pollset* ps, gpr_mu** mu) {
*mu = &ps->mu;
}
-static void pollset_destroy(grpc_pollset* ps) { gpr_mu_destroy(&ps->mu); }
+static void pollset_destroy(grpc_exec_ctx* exec_ctx, grpc_pollset* ps) {
+ gpr_mu_destroy(&ps->mu);
+}
static grpc_error* pollset_kick(grpc_pollset* p, grpc_pollset_worker* worker) {
return GRPC_ERROR_NONE;
diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
index 47705d3031..01ff39121e 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
@@ -105,6 +105,17 @@ static void BM_PumpStreamClientToServer(benchmark::State& state) {
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
+ response_rw.Finish(Status::OK, tag(0));
+ Status final_status;
+ request_rw->Finish(&final_status, tag(1));
+ need_tags = (1 << 0) | (1 << 1);
+ while (need_tags) {
+ GPR_ASSERT(fixture->cq()->Next(&t, &ok));
+ int i = (int)(intptr_t)t;
+ GPR_ASSERT(need_tags & (1 << i));
+ need_tags &= ~(1 << i);
+ }
+ GPR_ASSERT(final_status.ok());
}
fixture->Finish(state);
fixture.reset();
diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index fc99b06dbb..6b9fa8b38d 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -53,7 +53,7 @@ DEFINE_int32(
"Number of megabytes to pump before collecting flow control stats");
DEFINE_int32(
warmup_iterations, 100,
- "Number of megabytes to pump before collecting flow control stats");
+ "Number of iterations to run before collecting flow control stats");
DEFINE_int32(warmup_max_time_seconds, 10,
"Maximum number of seconds to run warmup loop");
@@ -77,13 +77,14 @@ static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
class TrickledCHTTP2 : public EndpointPairFixture {
public:
- TrickledCHTTP2(Service* service, size_t message_size,
- size_t kilobits_per_second)
+ TrickledCHTTP2(Service* service, bool streaming, size_t req_size,
+ size_t resp_size, size_t kilobits_per_second)
: EndpointPairFixture(service, MakeEndpoints(kilobits_per_second),
FixtureConfiguration()) {
if (FLAGS_log) {
std::ostringstream fn;
- fn << "trickle." << message_size << "." << kilobits_per_second << ".csv";
+ fn << "trickle." << (streaming ? "streaming" : "unary") << "." << req_size
+ << "." << resp_size << "." << kilobits_per_second << ".csv";
log_.reset(new std::ofstream(fn.str().c_str()));
write_csv(log_.get(), "t", "iteration", "client_backlog",
"server_backlog", "client_t_stall", "client_s_stall",
@@ -242,8 +243,9 @@ static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
EchoTestService::AsyncService service;
- std::unique_ptr<TrickledCHTTP2> fixture(
- new TrickledCHTTP2(&service, state.range(0), state.range(1)));
+ std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
+ &service, true, state.range(0) /* req_size */,
+ state.range(0) /* resp_size */, state.range(1) /* bw in kbit/s */));
{
EchoResponse send_response;
EchoResponse recv_response;
@@ -314,11 +316,7 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
state.SetBytesProcessed(state.range(0) * state.iterations());
}
-/*******************************************************************************
- * CONFIGURATIONS
- */
-
-static void TrickleArgs(benchmark::internal::Benchmark* b) {
+static void StreamingTrickleArgs(benchmark::internal::Benchmark* b) {
for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
for (int j = 64; j <= 128 * 1024 * 1024; j *= 8) {
double expected_time =
@@ -328,8 +326,111 @@ static void TrickleArgs(benchmark::internal::Benchmark* b) {
}
}
}
+BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(StreamingTrickleArgs);
-BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
+static void BM_PumpUnbalancedUnary_Trickle(benchmark::State& state) {
+ EchoTestService::AsyncService service;
+ std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
+ &service, true, state.range(0) /* req_size */,
+ state.range(1) /* resp_size */, state.range(2) /* bw in kbit/s */));
+ EchoRequest send_request;
+ EchoResponse send_response;
+ EchoResponse recv_response;
+ if (state.range(0) > 0) {
+ send_request.set_message(std::string(state.range(0), 'a'));
+ }
+ if (state.range(1) > 0) {
+ send_response.set_message(std::string(state.range(1), 'a'));
+ }
+ Status recv_status;
+ struct ServerEnv {
+ ServerContext ctx;
+ EchoRequest recv_request;
+ grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
+ ServerEnv() : response_writer(&ctx) {}
+ };
+ uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
+ ServerEnv* server_env[2] = {
+ reinterpret_cast<ServerEnv*>(server_env_buffer),
+ reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
+ new (server_env[0]) ServerEnv;
+ new (server_env[1]) ServerEnv;
+ service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
+ &server_env[0]->response_writer, fixture->cq(),
+ fixture->cq(), tag(0));
+ service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
+ &server_env[1]->response_writer, fixture->cq(),
+ fixture->cq(), tag(1));
+ std::unique_ptr<EchoTestService::Stub> stub(
+ EchoTestService::NewStub(fixture->channel()));
+ auto inner_loop = [&](bool in_warmup) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
+ recv_response.Clear();
+ ClientContext cli_ctx;
+ std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
+ stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
+ void* t;
+ bool ok;
+ TrickleCQNext(fixture.get(), &t, &ok, state.iterations());
+ GPR_ASSERT(ok);
+ GPR_ASSERT(t == tag(0) || t == tag(1));
+ intptr_t slot = reinterpret_cast<intptr_t>(t);
+ ServerEnv* senv = server_env[slot];
+ senv->response_writer.Finish(send_response, Status::OK, tag(3));
+ response_reader->Finish(&recv_response, &recv_status, tag(4));
+ for (int i = (1 << 3) | (1 << 4); i != 0;) {
+ TrickleCQNext(fixture.get(), &t, &ok, state.iterations());
+ GPR_ASSERT(ok);
+ int tagnum = (int)reinterpret_cast<intptr_t>(t);
+ GPR_ASSERT(i & (1 << tagnum));
+ i -= 1 << tagnum;
+ }
+ GPR_ASSERT(recv_status.ok());
+
+ senv->~ServerEnv();
+ senv = new (senv) ServerEnv();
+ service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
+ fixture->cq(), fixture->cq(), tag(slot));
+ };
+ gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
+ for (int i = 0;
+ i < GPR_MAX(FLAGS_warmup_iterations, FLAGS_warmup_megabytes * 1024 *
+ 1024 / (14 + state.range(0)));
+ i++) {
+ inner_loop(true);
+ if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
+ gpr_time_from_seconds(FLAGS_warmup_max_time_seconds,
+ GPR_TIMESPAN)) > 0) {
+ break;
+ }
+ }
+ while (state.KeepRunning()) {
+ inner_loop(false);
+ }
+ fixture->Finish(state);
+ fixture.reset();
+ server_env[0]->~ServerEnv();
+ server_env[1]->~ServerEnv();
+ state.SetBytesProcessed(state.range(0) * state.iterations() +
+ state.range(1) * state.iterations());
+}
+
+static void UnaryTrickleArgs(benchmark::internal::Benchmark* b) {
+ const int cli_1024k = 1024 * 1024;
+ const int cli_32M = 32 * 1024 * 1024;
+ const int svr_256k = 256 * 1024;
+ const int svr_4M = 4 * 1024 * 1024;
+ const int svr_64M = 64 * 1024 * 1024;
+ for (int bw = 64; bw <= 128 * 1024 * 1024; bw *= 16) {
+ b->Args({bw, cli_1024k, svr_256k});
+ b->Args({bw, cli_1024k, svr_4M});
+ b->Args({bw, cli_1024k, svr_64M});
+ b->Args({bw, cli_32M, svr_256k});
+ b->Args({bw, cli_32M, svr_4M});
+ b->Args({bw, cli_32M, svr_64M});
+ }
+}
+BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
}
}
diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc
index 0f3d3cef66..f5e8d13881 100644
--- a/test/cpp/microbenchmarks/bm_pollset.cc
+++ b/test/cpp/microbenchmarks/bm_pollset.cc
@@ -59,7 +59,7 @@ extern "C" {
auto& force_library_initialization = Library::get();
static void shutdown_ps(grpc_exec_ctx* exec_ctx, void* ps, grpc_error* error) {
- grpc_pollset_destroy(static_cast<grpc_pollset*>(ps));
+ grpc_pollset_destroy(exec_ctx, static_cast<grpc_pollset*>(ps));
}
static void BM_CreateDestroyPollset(benchmark::State& state) {
@@ -136,8 +136,7 @@ static void BM_PollEmptyPollset(benchmark::State& state) {
gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC);
gpr_mu_lock(mu);
while (state.KeepRunning()) {
- grpc_pollset_worker* worker;
- GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline));
+ GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline));
}
grpc_closure shutdown_ps_closure;
grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps,
@@ -150,6 +149,34 @@ static void BM_PollEmptyPollset(benchmark::State& state) {
}
BENCHMARK(BM_PollEmptyPollset);
+static void BM_PollAddFd(benchmark::State& state) {
+ TrackCounters track_counters;
+ size_t ps_sz = grpc_pollset_size();
+ grpc_pollset* ps = static_cast<grpc_pollset*>(gpr_zalloc(ps_sz));
+ gpr_mu* mu;
+ grpc_pollset_init(ps, &mu);
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ grpc_wakeup_fd wakeup_fd;
+ GPR_ASSERT(
+ GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&wakeup_fd)));
+ grpc_fd* fd = grpc_fd_create(wakeup_fd.read_fd, "xxx");
+ while (state.KeepRunning()) {
+ grpc_pollset_add_fd(&exec_ctx, ps, fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+ }
+ grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, "xxx");
+ grpc_closure shutdown_ps_closure;
+ grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps,
+ grpc_schedule_on_exec_ctx);
+ gpr_mu_lock(mu);
+ grpc_pollset_shutdown(&exec_ctx, ps, &shutdown_ps_closure);
+ gpr_mu_unlock(mu);
+ grpc_exec_ctx_finish(&exec_ctx);
+ gpr_free(ps);
+ track_counters.Finish(state);
+}
+BENCHMARK(BM_PollAddFd);
+
class Closure : public grpc_closure {
public:
virtual ~Closure() {}
@@ -233,8 +260,7 @@ static void BM_SingleThreadPollOneFd(benchmark::State& state) {
grpc_fd_notify_on_read(&exec_ctx, wakeup, continue_closure);
gpr_mu_lock(mu);
while (!done) {
- grpc_pollset_worker* worker;
- GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline));
+ GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline));
}
grpc_fd_orphan(&exec_ctx, wakeup, NULL, NULL, "done");
wakeup_fd.read_fd = 0;
diff --git a/test/cpp/performance/writes_per_rpc_test.cc b/test/cpp/performance/writes_per_rpc_test.cc
index 7a914c1547..12d8268330 100644
--- a/test/cpp/performance/writes_per_rpc_test.cc
+++ b/test/cpp/performance/writes_per_rpc_test.cc
@@ -254,8 +254,8 @@ TEST(WritesPerRpcTest, UnaryPingPong) {
EXPECT_LT(UnaryPingPong(0, 0), 2.05);
EXPECT_LT(UnaryPingPong(1, 0), 2.05);
EXPECT_LT(UnaryPingPong(0, 1), 2.05);
- EXPECT_LT(UnaryPingPong(4096, 0), 2.2);
- EXPECT_LT(UnaryPingPong(0, 4096), 2.2);
+ EXPECT_LT(UnaryPingPong(4096, 0), 2.5);
+ EXPECT_LT(UnaryPingPong(0, 4096), 2.5);
}
} // namespace testing
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index b94302d2f4..6b8f736813 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -246,39 +246,27 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
void* got_tag;
bool ok;
- switch (cli_cqs_[thread_idx]->AsyncNext(
- &got_tag, &ok,
- std::chrono::system_clock::now() + std::chrono::milliseconds(10))) {
- case CompletionQueue::GOT_EVENT: {
- // Got a regular event, so process it
- ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
- // Proceed while holding a lock to make sure that
- // this thread isn't supposed to shut down
- std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
- if (shutdown_state_[thread_idx]->shutdown) {
- delete ctx;
- return true;
- } else if (!ctx->RunNextState(ok, entry)) {
- // The RPC and callback are done, so clone the ctx
- // and kickstart the new one
- ctx->StartNewClone(cli_cqs_[thread_idx].get());
- // delete the old version
- delete ctx;
- }
+ if (cli_cqs_[thread_idx]->Next(&got_tag, &ok)) {
+ // Got a regular event, so process it
+ ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
+ // Proceed while holding a lock to make sure that
+ // this thread isn't supposed to shut down
+ std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
+ if (shutdown_state_[thread_idx]->shutdown) {
+ delete ctx;
return true;
+ } else if (!ctx->RunNextState(ok, entry)) {
+ // The RPC and callback are done, so clone the ctx
+ // and kickstart the new one
+ ctx->StartNewClone(cli_cqs_[thread_idx].get());
+ // delete the old version
+ delete ctx;
}
- case CompletionQueue::TIMEOUT: {
- std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
- if (shutdown_state_[thread_idx]->shutdown) {
- return true;
- }
- return true;
- }
- case CompletionQueue::SHUTDOWN: // queue is shutting down, so we must be
- // done
- return true;
+ return true;
+ } else {
+ // queue is shutting down, so we must be done
+ return true;
}
- GPR_UNREACHABLE_CODE(return true);
}
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-19 17:20:36 +0000