|
|
|
|
@ -34,6 +34,8 @@ |
|
|
|
|
/* Benchmark gRPC end2end in various configurations */ |
|
|
|
|
|
|
|
|
|
#include <benchmark/benchmark.h> |
|
|
|
|
#include <gflags/gflags.h> |
|
|
|
|
#include <fstream> |
|
|
|
|
#include "src/core/lib/profiling/timers.h" |
|
|
|
|
#include "src/cpp/client/create_channel_internal.h" |
|
|
|
|
#include "src/proto/grpc/testing/echo.grpc.pb.h" |
|
|
|
|
@ -45,16 +47,57 @@ extern "C" { |
|
|
|
|
#include "test/core/util/trickle_endpoint.h" |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
DEFINE_bool(log, false, "Log state to CSV files"); |
|
|
|
|
DEFINE_int32( |
|
|
|
|
warmup_megabytes, 1, |
|
|
|
|
"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"); |
|
|
|
|
DEFINE_int32(warmup_max_time_seconds, 10, |
|
|
|
|
"Maximum number of seconds to run warmup loop"); |
|
|
|
|
|
|
|
|
|
namespace grpc { |
|
|
|
|
namespace testing { |
|
|
|
|
|
|
|
|
|
static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); } |
|
|
|
|
|
|
|
|
|
template <class A0> |
|
|
|
|
static void write_csv(std::ostream* out, A0&& a0) { |
|
|
|
|
if (!out) return; |
|
|
|
|
(*out) << a0 << "\n"; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <class A0, class... Arg> |
|
|
|
|
static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) { |
|
|
|
|
if (!out) return; |
|
|
|
|
(*out) << a0 << ","; |
|
|
|
|
write_csv(out, std::forward<Arg>(arg)...); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
class TrickledCHTTP2 : public EndpointPairFixture { |
|
|
|
|
public: |
|
|
|
|
TrickledCHTTP2(Service* service, size_t megabits_per_second) |
|
|
|
|
: EndpointPairFixture(service, MakeEndpoints(megabits_per_second), |
|
|
|
|
FixtureConfiguration()) {} |
|
|
|
|
TrickledCHTTP2(Service* service, size_t message_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"; |
|
|
|
|
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", |
|
|
|
|
"server_t_stall", "server_s_stall", "client_t_outgoing", |
|
|
|
|
"server_t_outgoing", "client_t_incoming", "server_t_incoming", |
|
|
|
|
"client_s_outgoing_delta", "server_s_outgoing_delta", |
|
|
|
|
"client_s_incoming_delta", "server_s_incoming_delta", |
|
|
|
|
"client_s_announce_window", "server_s_announce_window", |
|
|
|
|
"client_peer_iws", "client_local_iws", "client_sent_iws", |
|
|
|
|
"client_acked_iws", "server_peer_iws", "server_local_iws", |
|
|
|
|
"server_sent_iws", "server_acked_iws", "client_queued_bytes", |
|
|
|
|
"server_queued_bytes"); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void AddToLabel(std::ostream& out, benchmark::State& state) { |
|
|
|
|
out << " writes/iter:" |
|
|
|
|
@ -75,7 +118,58 @@ class TrickledCHTTP2 : public EndpointPairFixture { |
|
|
|
|
(double)state.iterations()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Step() { |
|
|
|
|
void Log(int64_t iteration) { |
|
|
|
|
auto now = gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), start_); |
|
|
|
|
grpc_chttp2_transport* client = |
|
|
|
|
reinterpret_cast<grpc_chttp2_transport*>(client_transport_); |
|
|
|
|
grpc_chttp2_transport* server = |
|
|
|
|
reinterpret_cast<grpc_chttp2_transport*>(server_transport_); |
|
|
|
|
grpc_chttp2_stream* client_stream = |
|
|
|
|
client->stream_map.count == 1 |
|
|
|
|
? static_cast<grpc_chttp2_stream*>(client->stream_map.values[0]) |
|
|
|
|
: nullptr; |
|
|
|
|
grpc_chttp2_stream* server_stream = |
|
|
|
|
server->stream_map.count == 1 |
|
|
|
|
? static_cast<grpc_chttp2_stream*>(server->stream_map.values[0]) |
|
|
|
|
: nullptr; |
|
|
|
|
write_csv( |
|
|
|
|
log_.get(), static_cast<double>(now.tv_sec) + |
|
|
|
|
1e-9 * static_cast<double>(now.tv_nsec), |
|
|
|
|
iteration, grpc_trickle_get_backlog(endpoint_pair_.client), |
|
|
|
|
grpc_trickle_get_backlog(endpoint_pair_.server), |
|
|
|
|
client->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr, |
|
|
|
|
client->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr, |
|
|
|
|
server->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr, |
|
|
|
|
server->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr, |
|
|
|
|
client->outgoing_window, server->outgoing_window, |
|
|
|
|
client->incoming_window, server->incoming_window, |
|
|
|
|
client_stream ? client_stream->outgoing_window_delta : -1, |
|
|
|
|
server_stream ? server_stream->outgoing_window_delta : -1, |
|
|
|
|
client_stream ? client_stream->incoming_window_delta : -1, |
|
|
|
|
server_stream ? server_stream->incoming_window_delta : -1, |
|
|
|
|
client_stream ? client_stream->announce_window : -1, |
|
|
|
|
server_stream ? server_stream->announce_window : -1, |
|
|
|
|
client->settings[GRPC_PEER_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
client->settings[GRPC_LOCAL_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
client->settings[GRPC_SENT_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
client->settings[GRPC_ACKED_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
server->settings[GRPC_PEER_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
server->settings[GRPC_LOCAL_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
server->settings[GRPC_SENT_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
server->settings[GRPC_ACKED_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
client_stream ? client_stream->flow_controlled_buffer.length : 0, |
|
|
|
|
server_stream ? server_stream->flow_controlled_buffer.length : 0); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Step(bool update_stats) { |
|
|
|
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
|
|
|
|
size_t client_backlog = |
|
|
|
|
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.client); |
|
|
|
|
@ -83,10 +177,12 @@ class TrickledCHTTP2 : public EndpointPairFixture { |
|
|
|
|
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); |
|
|
|
|
if (update_stats) { |
|
|
|
|
UpdateStats((grpc_chttp2_transport*)client_transport_, &client_stats_, |
|
|
|
|
client_backlog); |
|
|
|
|
UpdateStats((grpc_chttp2_transport*)server_transport_, &server_stats_, |
|
|
|
|
server_backlog); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
@ -97,6 +193,8 @@ class TrickledCHTTP2 : public EndpointPairFixture { |
|
|
|
|
}; |
|
|
|
|
Stats client_stats_; |
|
|
|
|
Stats server_stats_; |
|
|
|
|
std::unique_ptr<std::ofstream> log_; |
|
|
|
|
gpr_timespec start_ = gpr_now(GPR_CLOCK_MONOTONIC); |
|
|
|
|
|
|
|
|
|
grpc_endpoint_pair MakeEndpoints(size_t kilobits) { |
|
|
|
|
grpc_endpoint_pair p; |
|
|
|
|
@ -123,13 +221,15 @@ class TrickledCHTTP2 : public EndpointPairFixture { |
|
|
|
|
// force library initialization
|
|
|
|
|
auto& force_library_initialization = Library::get(); |
|
|
|
|
|
|
|
|
|
static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok) { |
|
|
|
|
static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok, |
|
|
|
|
int64_t iteration) { |
|
|
|
|
while (true) { |
|
|
|
|
fixture->Log(iteration); |
|
|
|
|
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(); |
|
|
|
|
fixture->Step(iteration != -1); |
|
|
|
|
break; |
|
|
|
|
case CompletionQueue::SHUTDOWN: |
|
|
|
|
GPR_ASSERT(false); |
|
|
|
|
@ -143,7 +243,7 @@ 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(1))); |
|
|
|
|
new TrickledCHTTP2(&service, state.range(0), state.range(1))); |
|
|
|
|
{ |
|
|
|
|
EchoResponse send_response; |
|
|
|
|
EchoResponse recv_response; |
|
|
|
|
@ -163,18 +263,19 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) { |
|
|
|
|
void* t; |
|
|
|
|
bool ok; |
|
|
|
|
while (need_tags) { |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok); |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok, -1); |
|
|
|
|
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()) { |
|
|
|
|
auto inner_loop = [&](bool in_warmup) { |
|
|
|
|
GPR_TIMER_SCOPE("BenchmarkCycle", 0); |
|
|
|
|
response_rw.Write(send_response, tag(1)); |
|
|
|
|
while (true) { |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok); |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok, |
|
|
|
|
in_warmup ? -1 : state.iterations()); |
|
|
|
|
if (t == tag(0)) { |
|
|
|
|
request_rw->Read(&recv_response, tag(0)); |
|
|
|
|
} else if (t == tag(1)) { |
|
|
|
|
@ -183,11 +284,26 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) { |
|
|
|
|
GPR_ASSERT(false); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
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); |
|
|
|
|
} |
|
|
|
|
response_rw.Finish(Status::OK, tag(1)); |
|
|
|
|
need_tags = (1 << 0) | (1 << 1); |
|
|
|
|
while (need_tags) { |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok); |
|
|
|
|
TrickleCQNext(fixture.get(), &t, &ok, -1); |
|
|
|
|
int i = (int)(intptr_t)t; |
|
|
|
|
GPR_ASSERT(need_tags & (1 << i)); |
|
|
|
|
need_tags &= ~(1 << i); |
|
|
|
|
@ -204,10 +320,10 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) { |
|
|
|
|
|
|
|
|
|
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) { |
|
|
|
|
for (int j = 64; 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; |
|
|
|
|
if (expected_time > 2.0) continue; |
|
|
|
|
b->Args({i, j}); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -217,4 +333,8 @@ BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
BENCHMARK_MAIN(); |
|
|
|
|
int main(int argc, char** argv) { |
|
|
|
|
::benchmark::Initialize(&argc, argv); |
|
|
|
|
::google::ParseCommandLineFlags(&argc, &argv, false); |
|
|
|
|
::benchmark::RunSpecifiedBenchmarks(); |
|
|
|
|
} |
|
|
|
|
|
Craig Tiller
8 years ago
committed by
GitHub