|
|
|
/*
|
|
|
|
*
|
|
|
|
* Copyright 2016, 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.
|
|
|
|
*
|
|
|
|
*/
|
|
|
|
|
|
|
|
/* 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"
|
|
|
|
#include "test/cpp/microbenchmarks/fullstack_context_mutators.h"
|
|
|
|
#include "test/cpp/microbenchmarks/fullstack_fixtures.h"
|
|
|
|
extern "C" {
|
|
|
|
#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
|
|
|
|
#include "src/core/ext/transport/chttp2/transport/internal.h"
|
|
|
|
#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 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:"
|
|
|
|
<< ((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 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);
|
|
|
|
size_t server_backlog =
|
|
|
|
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.server);
|
|
|
|
grpc_exec_ctx_finish(&exec_ctx);
|
|
|
|
|
|
|
|
if (update_stats) {
|
|
|
|
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_;
|
|
|
|
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;
|
|
|
|
grpc_passthru_endpoint_create(&p.client, &p.server, Library::get().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++;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
// force library initialization
|
|
|
|
auto& force_library_initialization = Library::get();
|
|
|
|
|
|
|
|
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(iteration != -1);
|
|
|
|
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(0), 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, -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));
|
|
|
|
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,
|
|
|
|
in_warmup ? -1 : state.iterations());
|
|
|
|
if (t == tag(0)) {
|
|
|
|
request_rw->Read(&recv_response, tag(0));
|
|
|
|
} else if (t == tag(1)) {
|
|
|
|
break;
|
|
|
|
} else {
|
|
|
|
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, -1);
|
|
|
|
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
|
|
|
|
*/
|
|
|
|
|
|
|
|
static void TrickleArgs(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 =
|
|
|
|
static_cast<double>(14 + i) / (125.0 * static_cast<double>(j));
|
|
|
|
if (expected_time > 2.0) continue;
|
|
|
|
b->Args({i, j});
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
int main(int argc, char** argv) {
|
|
|
|
::benchmark::Initialize(&argc, argv);
|
|
|
|
::google::ParseCommandLineFlags(&argc, &argv, false);
|
|
|
|
::benchmark::RunSpecifiedBenchmarks();
|
|
|
|
}
|