aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/microbenchmarks/bm_fullstack.cc
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/microbenchmarks/bm_fullstack.cc')
-rw-r--r--test/cpp/microbenchmarks/bm_fullstack.cc244
1 files changed, 225 insertions, 19 deletions
diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc
index dd180de48a..c63de0ce0a 100644
--- a/test/cpp/microbenchmarks/bm_fullstack.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack.cc
@@ -46,6 +46,7 @@
extern "C" {
#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/ext/transport/chttp2/transport/internal.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/endpoint_pair.h"
@@ -54,9 +55,12 @@ extern "C" {
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/completion_queue.h"
#include "src/core/lib/surface/server.h"
+#include "test/core/util/memory_counters.h"
#include "test/core/util/passthru_endpoint.h"
#include "test/core/util/port.h"
+#include "test/core/util/trickle_endpoint.h"
}
+#include "src/core/lib/profiling/timers.h"
#include "src/cpp/client/create_channel_internal.h"
#include "src/proto/grpc/testing/echo.grpc.pb.h"
#include "third_party/benchmark/include/benchmark/benchmark.h"
@@ -67,6 +71,7 @@ namespace testing {
static class InitializeStuff {
public:
InitializeStuff() {
+ grpc_memory_counters_init();
init_lib_.init();
rq_ = grpc_resource_quota_create("bm");
}
@@ -94,7 +99,42 @@ static void ApplyCommonChannelArguments(ChannelArguments* c) {
c->SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, INT_MAX);
}
-class FullstackFixture {
+#ifdef GPR_MU_COUNTERS
+extern "C" gpr_atm grpc_mu_locks;
+#endif
+
+class BaseFixture {
+ public:
+ void Finish(benchmark::State& s) {
+ std::ostringstream out;
+ this->AddToLabel(out, s);
+#ifdef GPR_MU_COUNTERS
+ out << " locks/iter:" << ((double)(gpr_atm_no_barrier_load(&grpc_mu_locks) -
+ mu_locks_at_start_) /
+ (double)s.iterations());
+#endif
+ grpc_memory_counters counters_at_end = grpc_memory_counters_snapshot();
+ out << " allocs/iter:"
+ << ((double)(counters_at_end.total_allocs_absolute -
+ counters_at_start_.total_allocs_absolute) /
+ (double)s.iterations());
+ auto label = out.str();
+ if (label.length() && label[0] == ' ') {
+ label = label.substr(1);
+ }
+ s.SetLabel(label);
+ }
+
+ virtual void AddToLabel(std::ostream& out, benchmark::State& s) = 0;
+
+ private:
+#ifdef GPR_MU_COUNTERS
+ const size_t mu_locks_at_start_ = gpr_atm_no_barrier_load(&grpc_mu_locks);
+#endif
+ grpc_memory_counters counters_at_start_ = grpc_memory_counters_snapshot();
+};
+
+class FullstackFixture : public BaseFixture {
public:
FullstackFixture(Service* service, const grpc::string& address) {
ServerBuilder b;
@@ -130,7 +170,7 @@ class TCP : public FullstackFixture {
public:
TCP(Service* service) : FullstackFixture(service, MakeAddress()) {}
- void Finish(benchmark::State& state) {}
+ void AddToLabel(std::ostream& out, benchmark::State& state) {}
private:
static grpc::string MakeAddress() {
@@ -145,7 +185,7 @@ class UDS : public FullstackFixture {
public:
UDS(Service* service) : FullstackFixture(service, MakeAddress()) {}
- void Finish(benchmark::State& state) {}
+ void AddToLabel(std::ostream& out, benchmark::State& state) override {}
private:
static grpc::string MakeAddress() {
@@ -157,9 +197,10 @@ class UDS : public FullstackFixture {
}
};
-class EndpointPairFixture {
+class EndpointPairFixture : public BaseFixture {
public:
- EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints) {
+ EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints)
+ : endpoint_pair_(endpoints) {
ServerBuilder b;
cq_ = b.AddCompletionQueue(true);
b.RegisterService(service);
@@ -172,7 +213,7 @@ class EndpointPairFixture {
{
const grpc_channel_args* server_args =
grpc_server_get_channel_args(server_->c_server());
- grpc_transport* transport = grpc_create_chttp2_transport(
+ server_transport_ = grpc_create_chttp2_transport(
&exec_ctx, server_args, endpoints.server, 0 /* is_client */);
grpc_pollset** pollsets;
@@ -183,9 +224,9 @@ class EndpointPairFixture {
grpc_endpoint_add_to_pollset(&exec_ctx, endpoints.server, pollsets[i]);
}
- grpc_server_setup_transport(&exec_ctx, server_->c_server(), transport,
- NULL, server_args);
- grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+ grpc_server_setup_transport(&exec_ctx, server_->c_server(),
+ server_transport_, NULL, server_args);
+ grpc_chttp2_transport_start_reading(&exec_ctx, server_transport_, NULL);
}
/* create channel */
@@ -195,12 +236,13 @@ class EndpointPairFixture {
ApplyCommonChannelArguments(&args);
grpc_channel_args c_args = args.c_channel_args();
- grpc_transport* transport =
+ client_transport_ =
grpc_create_chttp2_transport(&exec_ctx, &c_args, endpoints.client, 1);
- GPR_ASSERT(transport);
- grpc_channel* channel = grpc_channel_create(
- &exec_ctx, "target", &c_args, GRPC_CLIENT_DIRECT_CHANNEL, transport);
- grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
+ GPR_ASSERT(client_transport_);
+ grpc_channel* channel =
+ grpc_channel_create(&exec_ctx, "target", &c_args,
+ GRPC_CLIENT_DIRECT_CHANNEL, client_transport_);
+ grpc_chttp2_transport_start_reading(&exec_ctx, client_transport_, NULL);
channel_ = CreateChannelInternal("", channel);
}
@@ -220,6 +262,11 @@ class EndpointPairFixture {
ServerCompletionQueue* cq() { return cq_.get(); }
std::shared_ptr<Channel> channel() { return channel_; }
+ protected:
+ grpc_endpoint_pair endpoint_pair_;
+ grpc_transport* client_transport_;
+ grpc_transport* server_transport_;
+
private:
std::unique_ptr<Server> server_;
std::unique_ptr<ServerCompletionQueue> cq_;
@@ -233,7 +280,7 @@ class SockPair : public EndpointPairFixture {
"test", initialize_stuff.rq(), 8192)) {
}
- void Finish(benchmark::State& state) {}
+ void AddToLabel(std::ostream& out, benchmark::State& state) {}
};
class InProcessCHTTP2 : public EndpointPairFixture {
@@ -241,11 +288,9 @@ class InProcessCHTTP2 : public EndpointPairFixture {
InProcessCHTTP2(Service* service)
: EndpointPairFixture(service, MakeEndpoints()) {}
- void Finish(benchmark::State& state) {
- std::ostringstream out;
- out << "writes/iteration:"
+ void AddToLabel(std::ostream& out, benchmark::State& state) {
+ out << " writes/iter:"
<< ((double)stats_.num_writes / (double)state.iterations());
- state.SetLabel(out.str());
}
private:
@@ -259,6 +304,75 @@ class InProcessCHTTP2 : public EndpointPairFixture {
}
};
+class TrickledCHTTP2 : public EndpointPairFixture {
+ public:
+ TrickledCHTTP2(Service* service, size_t megabits_per_second)
+ : EndpointPairFixture(service, MakeEndpoints(megabits_per_second)) {}
+
+ void AddToLabel(std::ostream& out, benchmark::State& state) {
+ out << " writes/iter:"
+ << ((double)stats_.num_writes / (double)state.iterations())
+ << " cli_transport_stalls/iter:"
+ << ((double)
+ client_stats_.streams_stalled_due_to_transport_flow_control /
+ (double)state.iterations())
+ << " cli_stream_stalls/iter:"
+ << ((double)client_stats_.streams_stalled_due_to_stream_flow_control /
+ (double)state.iterations())
+ << " svr_transport_stalls/iter:"
+ << ((double)
+ server_stats_.streams_stalled_due_to_transport_flow_control /
+ (double)state.iterations())
+ << " svr_stream_stalls/iter:"
+ << ((double)server_stats_.streams_stalled_due_to_stream_flow_control /
+ (double)state.iterations());
+ }
+
+ void Step() {
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ size_t client_backlog =
+ grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.client);
+ size_t server_backlog =
+ grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.server);
+ grpc_exec_ctx_finish(&exec_ctx);
+
+ UpdateStats((grpc_chttp2_transport*)client_transport_, &client_stats_,
+ client_backlog);
+ UpdateStats((grpc_chttp2_transport*)server_transport_, &server_stats_,
+ server_backlog);
+ }
+
+ private:
+ grpc_passthru_endpoint_stats stats_;
+ struct Stats {
+ int streams_stalled_due_to_stream_flow_control = 0;
+ int streams_stalled_due_to_transport_flow_control = 0;
+ };
+ Stats client_stats_;
+ Stats server_stats_;
+
+ grpc_endpoint_pair MakeEndpoints(size_t kilobits) {
+ grpc_endpoint_pair p;
+ grpc_passthru_endpoint_create(&p.client, &p.server, initialize_stuff.rq(),
+ &stats_);
+ double bytes_per_second = 125.0 * kilobits;
+ p.client = grpc_trickle_endpoint_create(p.client, bytes_per_second);
+ p.server = grpc_trickle_endpoint_create(p.server, bytes_per_second);
+ return p;
+ }
+
+ void UpdateStats(grpc_chttp2_transport* t, Stats* s, size_t backlog) {
+ if (backlog == 0) {
+ if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != NULL) {
+ s->streams_stalled_due_to_stream_flow_control++;
+ }
+ if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != NULL) {
+ s->streams_stalled_due_to_transport_flow_control++;
+ }
+ }
+ }
+};
+
/*******************************************************************************
* CONTEXT MUTATORS
*/
@@ -402,6 +516,7 @@ static void BM_UnaryPingPong(benchmark::State& state) {
std::unique_ptr<EchoTestService::Stub> stub(
EchoTestService::NewStub(fixture->channel()));
while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
recv_response.Clear();
ClientContext cli_ctx;
ClientContextMutator cli_ctx_mut(&cli_ctx);
@@ -583,6 +698,7 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) {
}
while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
request_rw->Write(send_request, tag(0)); // Start client send
response_rw.Read(&recv_request, tag(1)); // Start server recv
request_rw->Read(&recv_response, tag(2)); // Start client recv
@@ -655,6 +771,7 @@ static void BM_PumpStreamClientToServer(benchmark::State& state) {
}
response_rw.Read(&recv_request, tag(0));
while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
request_rw->Write(send_request, tag(1));
while (true) {
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
@@ -712,6 +829,7 @@ static void BM_PumpStreamServerToClient(benchmark::State& state) {
}
request_rw->Read(&recv_response, tag(0));
while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
response_rw.Write(send_response, tag(1));
while (true) {
GPR_ASSERT(fixture->cq()->Next(&t, &ok));
@@ -738,6 +856,81 @@ static void BM_PumpStreamServerToClient(benchmark::State& state) {
state.SetBytesProcessed(state.range(0) * state.iterations());
}
+static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok) {
+ while (true) {
+ switch (fixture->cq()->AsyncNext(
+ t, ok, gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
+ gpr_time_from_micros(100, GPR_TIMESPAN)))) {
+ case CompletionQueue::TIMEOUT:
+ fixture->Step();
+ break;
+ case CompletionQueue::SHUTDOWN:
+ GPR_ASSERT(false);
+ break;
+ case CompletionQueue::GOT_EVENT:
+ return;
+ }
+ }
+}
+
+static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
+ EchoTestService::AsyncService service;
+ std::unique_ptr<TrickledCHTTP2> fixture(
+ new TrickledCHTTP2(&service, state.range(1)));
+ {
+ EchoResponse send_response;
+ EchoResponse recv_response;
+ if (state.range(0) > 0) {
+ send_response.set_message(std::string(state.range(0), 'a'));
+ }
+ Status recv_status;
+ ServerContext svr_ctx;
+ ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
+ service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
+ fixture->cq(), tag(0));
+ std::unique_ptr<EchoTestService::Stub> stub(
+ EchoTestService::NewStub(fixture->channel()));
+ ClientContext cli_ctx;
+ auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
+ int need_tags = (1 << 0) | (1 << 1);
+ void* t;
+ bool ok;
+ while (need_tags) {
+ TrickleCQNext(fixture.get(), &t, &ok);
+ GPR_ASSERT(ok);
+ int i = (int)(intptr_t)t;
+ GPR_ASSERT(need_tags & (1 << i));
+ need_tags &= ~(1 << i);
+ }
+ request_rw->Read(&recv_response, tag(0));
+ while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
+ response_rw.Write(send_response, tag(1));
+ while (true) {
+ TrickleCQNext(fixture.get(), &t, &ok);
+ if (t == tag(0)) {
+ request_rw->Read(&recv_response, tag(0));
+ } else if (t == tag(1)) {
+ break;
+ } else {
+ GPR_ASSERT(false);
+ }
+ }
+ }
+ response_rw.Finish(Status::OK, tag(1));
+ need_tags = (1 << 0) | (1 << 1);
+ while (need_tags) {
+ TrickleCQNext(fixture.get(), &t, &ok);
+ int i = (int)(intptr_t)t;
+ GPR_ASSERT(need_tags & (1 << i));
+ need_tags &= ~(1 << i);
+ }
+ }
+ fixture->Finish(state);
+ fixture.reset();
+ state.SetBytesProcessed(state.range(0) * state.iterations());
+}
+
/*******************************************************************************
* CONFIGURATIONS
*/
@@ -827,6 +1020,19 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair)
BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2)
->Range(0, 128 * 1024 * 1024);
+static void TrickleArgs(benchmark::internal::Benchmark* b) {
+ for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
+ for (int j = 1; j <= 128 * 1024 * 1024; j *= 8) {
+ double expected_time =
+ static_cast<double>(14 + i) / (125.0 * static_cast<double>(j));
+ if (expected_time > 0.01) continue;
+ b->Args({i, j});
+ }
+ }
+}
+
+BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
+
// Generate Args for StreamingPingPong benchmarks. Currently generates args for
// only "small streams" (i.e streams with 0, 1 or 2 messages)
static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) {
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-06 15:03:04 +0000