Skip to content
Snippets Groups Projects
Commit b6ac9b45 authored by ncteisen's avatar ncteisen
Browse files

Add unbalanced unary benchmark

parent d64a4bea
No related branches found
No related tags found
No related merge requests found
......@@ -77,13 +77,17 @@ static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
class TrickledCHTTP2 : public EndpointPairFixture {
public:
TrickledCHTTP2(Service* service, size_t message_size,
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 +246,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 +319,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 +329,93 @@ 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()));
while (state.KeepRunning()) {
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));
}
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) {
for (int i = 1; i <= 128 * 1024 * 1024; i *= 32) {
for (int j = 1; j <= 128 * 1024 * 1024; j *= 32) {
for (int k = 64; k <= 128 * 1024 * 1024; k *= 16) {
double expected_time =
static_cast<double>(14 + i + k) / (125.0 * 2 * static_cast<double>(j));
if (expected_time > 2.0) continue;
b->Args({i, j, k});
}
}
}
}
BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
}
}
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment