test/cpp/microbenchmarks/bm_fullstack_trickle.cc

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 /* Benchmark gRPC end2end in various configurations */
  20
  21 #include <benchmark/benchmark.h>
  22
  23 #include <fstream>
  24
  25 #include "absl/flags/flag.h"
  26 #include "absl/memory/memory.h"
  27 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
  28 #include "src/core/ext/transport/chttp2/transport/internal.h"
  29 #include "src/core/lib/iomgr/timer_manager.h"
  30 #include "src/core/lib/profiling/timers.h"
  31 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  32 #include "test/core/util/test_config.h"
  33 #include "test/core/util/trickle_endpoint.h"
  34 #include "test/cpp/microbenchmarks/fullstack_context_mutators.h"
  35 #include "test/cpp/microbenchmarks/fullstack_fixtures.h"
  36 #include "test/cpp/util/test_config.h"
  37
  38 ABSL_FLAG(bool, log, false, "Log state to CSV files");
  39 ABSL_FLAG(int32_t, warmup_megabytes, 1,
  40           "Number of megabytes to pump before collecting flow control stats");
  41 ABSL_FLAG(int32_t, warmup_iterations, 100,
  42           "Number of iterations to run before collecting flow control stats");
  43 ABSL_FLAG(int32_t, warmup_max_time_seconds, 10,
  44           "Maximum number of seconds to run warmup loop");
  45
  46 namespace grpc {
  47 namespace testing {
  48
  49 gpr_atm g_now_us = 0;
  50
  51 static gpr_timespec fake_now(gpr_clock_type clock_type) {
  52   gpr_timespec t;
  53   gpr_atm now = gpr_atm_no_barrier_load(&g_now_us);
  54   t.tv_sec = now / GPR_US_PER_SEC;
  55   t.tv_nsec = (now % GPR_US_PER_SEC) * GPR_NS_PER_US;
  56   t.clock_type = clock_type;
  57   return t;
  58 }
  59
  60 static void inc_time() {
  61   gpr_atm_no_barrier_fetch_add(&g_now_us, 100);
  62   grpc_timer_manager_tick();
  63 }
  64
  65 static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
  66
  67 template <class A0>
  68 static void write_csv(std::ostream* out, A0&& a0) {
  69   if (!out) return;
  70   (*out) << a0 << "\n";
  71 }
  72
  73 template <class A0, class... Arg>
  74 static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
  75   if (!out) return;
  76   (*out) << a0 << ",";
  77   write_csv(out, std::forward<Arg>(arg)...);
  78 }
  79
  80 class TrickledCHTTP2 : public EndpointPairFixture {
  81  public:
  82   TrickledCHTTP2(Service* service, bool streaming, size_t req_size,
  83                  size_t resp_size, size_t kilobits_per_second,
  84                  grpc_passthru_endpoint_stats* stats)
  85       : EndpointPairFixture(service, MakeEndpoints(kilobits_per_second, stats),
  86                             FixtureConfiguration()),
  87         stats_(stats) {
  88     if (absl::GetFlag(FLAGS_log)) {
  89       std::ostringstream fn;
  90       fn << "trickle." << (streaming ? "streaming" : "unary") << "." << req_size
  91          << "." << resp_size << "." << kilobits_per_second << ".csv";
  92       log_ = absl::make_unique<std::ofstream>(fn.str().c_str());
  93       write_csv(log_.get(), "t", "iteration", "client_backlog",
  94                 "server_backlog", "client_t_stall", "client_s_stall",
  95                 "server_t_stall", "server_s_stall", "client_t_remote",
  96                 "server_t_remote", "client_t_announced", "server_t_announced",
  97                 "client_s_remote_delta", "server_s_remote_delta",
  98                 "client_s_local_delta", "server_s_local_delta",
  99                 "client_s_announced_delta", "server_s_announced_delta",
 100                 "client_peer_iws", "client_local_iws", "client_sent_iws",
 101                 "client_acked_iws", "server_peer_iws", "server_local_iws",
 102                 "server_sent_iws", "server_acked_iws", "client_queued_bytes",
 103                 "server_queued_bytes");
 104     }
 105   }
 106
 107   ~TrickledCHTTP2() override {
 108     if (stats_ != nullptr) {
 109       grpc_passthru_endpoint_stats_destroy(stats_);
 110     }
 111   }
 112
 113   void AddToLabel(std::ostream& out, benchmark::State& state) override {
 114     out << " writes/iter:"
 115         << (static_cast<double>(stats_->num_writes) /
 116             static_cast<double>(state.iterations()))
 117         << " cli_transport_stalls/iter:"
 118         << (static_cast<double>(
 119                 client_stats_.streams_stalled_due_to_transport_flow_control) /
 120             static_cast<double>(state.iterations()))
 121         << " cli_stream_stalls/iter:"
 122         << (static_cast<double>(
 123                 client_stats_.streams_stalled_due_to_stream_flow_control) /
 124             static_cast<double>(state.iterations()))
 125         << " svr_transport_stalls/iter:"
 126         << (static_cast<double>(
 127                 server_stats_.streams_stalled_due_to_transport_flow_control) /
 128             static_cast<double>(state.iterations()))
 129         << " svr_stream_stalls/iter:"
 130         << (static_cast<double>(
 131                 server_stats_.streams_stalled_due_to_stream_flow_control) /
 132             static_cast<double>(state.iterations()));
 133   }
 134
 135   void Log(int64_t iteration) GPR_ATTRIBUTE_NO_TSAN {
 136     auto now = gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), start_);
 137     grpc_chttp2_transport* client =
 138         reinterpret_cast<grpc_chttp2_transport*>(client_transport_);
 139     grpc_chttp2_transport* server =
 140         reinterpret_cast<grpc_chttp2_transport*>(server_transport_);
 141     grpc_chttp2_stream* client_stream =
 142         client->stream_map.count == 1
 143             ? static_cast<grpc_chttp2_stream*>(client->stream_map.values[0])
 144             : nullptr;
 145     grpc_chttp2_stream* server_stream =
 146         server->stream_map.count == 1
 147             ? static_cast<grpc_chttp2_stream*>(server->stream_map.values[0])
 148             : nullptr;
 149     write_csv(
 150         log_.get(),
 151         static_cast<double>(now.tv_sec) +
 152             1e-9 * static_cast<double>(now.tv_nsec),
 153         iteration, grpc_trickle_get_backlog(endpoint_pair_.client),
 154         grpc_trickle_get_backlog(endpoint_pair_.server),
 155         client->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr,
 156         client->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr,
 157         server->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr,
 158         server->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr,
 159         client->flow_control->remote_window_,
 160         server->flow_control->remote_window_,
 161         client->flow_control->announced_window_,
 162         server->flow_control->announced_window_,
 163         client_stream ? client_stream->flow_control->remote_window_delta_ : -1,
 164         server_stream ? server_stream->flow_control->remote_window_delta_ : -1,
 165         client_stream ? client_stream->flow_control->local_window_delta_ : -1,
 166         server_stream ? server_stream->flow_control->local_window_delta_ : -1,
 167         client_stream ? client_stream->flow_control->announced_window_delta_
 168                       : -1,
 169         server_stream ? server_stream->flow_control->announced_window_delta_
 170                       : -1,
 171         client->settings[GRPC_PEER_SETTINGS]
 172                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 173         client->settings[GRPC_LOCAL_SETTINGS]
 174                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 175         client->settings[GRPC_SENT_SETTINGS]
 176                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 177         client->settings[GRPC_ACKED_SETTINGS]
 178                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 179         server->settings[GRPC_PEER_SETTINGS]
 180                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 181         server->settings[GRPC_LOCAL_SETTINGS]
 182                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 183         server->settings[GRPC_SENT_SETTINGS]
 184                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 185         server->settings[GRPC_ACKED_SETTINGS]
 186                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
 187         client_stream ? client_stream->flow_controlled_buffer.length : 0,
 188         server_stream ? server_stream->flow_controlled_buffer.length : 0);
 189   }
 190
 191   void Step(bool update_stats) {
 192     grpc_core::ExecCtx exec_ctx;
 193     inc_time();
 194     size_t client_backlog =
 195         grpc_trickle_endpoint_trickle(endpoint_pair_.client);
 196     size_t server_backlog =
 197         grpc_trickle_endpoint_trickle(endpoint_pair_.server);
 198
 199     if (update_stats) {
 200       UpdateStats(reinterpret_cast<grpc_chttp2_transport*>(client_transport_),
 201                   &client_stats_, client_backlog);
 202       UpdateStats(reinterpret_cast<grpc_chttp2_transport*>(server_transport_),
 203                   &server_stats_, server_backlog);
 204     }
 205   }
 206
 207  private:
 208   grpc_passthru_endpoint_stats* stats_;
 209   struct Stats {
 210     int streams_stalled_due_to_stream_flow_control = 0;
 211     int streams_stalled_due_to_transport_flow_control = 0;
 212   };
 213   Stats client_stats_;
 214   Stats server_stats_;
 215   std::unique_ptr<std::ofstream> log_;
 216   gpr_timespec start_ = gpr_now(GPR_CLOCK_MONOTONIC);
 217
 218   static grpc_endpoint_pair MakeEndpoints(size_t kilobits,
 219                                           grpc_passthru_endpoint_stats* stats) {
 220     grpc_endpoint_pair p;
 221     grpc_passthru_endpoint_create(&p.client, &p.server,
 222                                   LibraryInitializer::get().rq(), stats);
 223     double bytes_per_second = 125.0 * kilobits;
 224     p.client = grpc_trickle_endpoint_create(p.client, bytes_per_second);
 225     p.server = grpc_trickle_endpoint_create(p.server, bytes_per_second);
 226     return p;
 227   }
 228
 229   void UpdateStats(grpc_chttp2_transport* t, Stats* s,
 230                    size_t backlog) GPR_ATTRIBUTE_NO_TSAN {
 231     if (backlog == 0) {
 232       if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr) {
 233         s->streams_stalled_due_to_stream_flow_control++;
 234       }
 235       if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr) {
 236         s->streams_stalled_due_to_transport_flow_control++;
 237       }
 238     }
 239   }
 240 };
 241
 242 static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
 243                           int64_t iteration) {
 244   while (true) {
 245     fixture->Log(iteration);
 246     switch (
 247         fixture->cq()->AsyncNext(t, ok, gpr_inf_past(GPR_CLOCK_MONOTONIC))) {
 248       case CompletionQueue::TIMEOUT:
 249         fixture->Step(iteration != -1);
 250         break;
 251       case CompletionQueue::SHUTDOWN:
 252         GPR_ASSERT(false);
 253         break;
 254       case CompletionQueue::GOT_EVENT:
 255         return;
 256     }
 257   }
 258 }
 259
 260 static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
 261   EchoTestService::AsyncService service;
 262   std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
 263       &service, true, state.range(0) /* req_size */,
 264       state.range(0) /* resp_size */, state.range(1) /* bw in kbit/s */,
 265       grpc_passthru_endpoint_stats_create()));
 266   {
 267     EchoResponse send_response;
 268     EchoResponse recv_response;
 269     if (state.range(0) > 0) {
 270       send_response.set_message(std::string(state.range(0), 'a'));
 271     }
 272     Status recv_status;
 273     ServerContext svr_ctx;
 274     ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
 275     service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
 276                               fixture->cq(), tag(0));
 277     std::unique_ptr<EchoTestService::Stub> stub(
 278         EchoTestService::NewStub(fixture->channel()));
 279     ClientContext cli_ctx;
 280     auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
 281     int need_tags = (1 << 0) | (1 << 1);
 282     void* t;
 283     bool ok;
 284     while (need_tags) {
 285       TrickleCQNext(fixture.get(), &t, &ok, -1);
 286       GPR_ASSERT(ok);
 287       int i = static_cast<int>(reinterpret_cast<intptr_t>(t));
 288       GPR_ASSERT(need_tags & (1 << i));
 289       need_tags &= ~(1 << i);
 290     }
 291     request_rw->Read(&recv_response, tag(0));
 292     auto inner_loop = [&](bool in_warmup) {
 293       GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 294       response_rw.Write(send_response, tag(1));
 295       while (true) {
 296         TrickleCQNext(fixture.get(), &t, &ok,
 297                       in_warmup ? -1 : state.iterations());
 298         if (t == tag(0)) {
 299           request_rw->Read(&recv_response, tag(0));
 300         } else if (t == tag(1)) {
 301           break;
 302         } else {
 303           GPR_ASSERT(false);
 304         }
 305       }
 306     };
 307     gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
 308     for (int i = 0; i < GPR_MAX(absl::GetFlag(FLAGS_warmup_iterations),
 309                                 absl::GetFlag(FLAGS_warmup_megabytes) * 1024 *
 310                                     1024 / (14 + state.range(0)));
 311          i++) {
 312       inner_loop(true);
 313       if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
 314                        gpr_time_from_seconds(
 315                            absl::GetFlag(FLAGS_warmup_max_time_seconds),
 316                            GPR_TIMESPAN)) > 0) {
 317         break;
 318       }
 319     }
 320     while (state.KeepRunning()) {
 321       inner_loop(false);
 322     }
 323     response_rw.Finish(Status::OK, tag(1));
 324     grpc::Status status;
 325     request_rw->Finish(&status, tag(2));
 326     need_tags = (1 << 0) | (1 << 1) | (1 << 2);
 327     while (need_tags) {
 328       TrickleCQNext(fixture.get(), &t, &ok, -1);
 329       if (t == tag(0) && ok) {
 330         request_rw->Read(&recv_response, tag(0));
 331         continue;
 332       }
 333       int i = static_cast<int>(reinterpret_cast<intptr_t>(t));
 334       GPR_ASSERT(need_tags & (1 << i));
 335       need_tags &= ~(1 << i);
 336     }
 337   }
 338   fixture->Finish(state);
 339   fixture.reset();
 340   state.SetBytesProcessed(state.range(0) * state.iterations());
 341 }
 342
 343 static void StreamingTrickleArgs(benchmark::internal::Benchmark* b) {
 344   for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
 345     for (int j = 64; j <= 128 * 1024 * 1024; j *= 8) {
 346       double expected_time =
 347           static_cast<double>(14 + i) / (125.0 * static_cast<double>(j));
 348       if (expected_time > 2.0) continue;
 349       b->Args({i, j});
 350     }
 351   }
 352 }
 353 BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(StreamingTrickleArgs);
 354
 355 static void BM_PumpUnbalancedUnary_Trickle(benchmark::State& state) {
 356   EchoTestService::AsyncService service;
 357   std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
 358       &service, false, state.range(0) /* req_size */,
 359       state.range(1) /* resp_size */, state.range(2) /* bw in kbit/s */,
 360       grpc_passthru_endpoint_stats_create()));
 361   EchoRequest send_request;
 362   EchoResponse send_response;
 363   EchoResponse recv_response;
 364   if (state.range(0) > 0) {
 365     send_request.set_message(std::string(state.range(0), 'a'));
 366   }
 367   if (state.range(1) > 0) {
 368     send_response.set_message(std::string(state.range(1), 'a'));
 369   }
 370   Status recv_status;
 371   struct ServerEnv {
 372     ServerContext ctx;
 373     EchoRequest recv_request;
 374     grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
 375     ServerEnv() : response_writer(&ctx) {}
 376   };
 377   uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
 378   ServerEnv* server_env[2] = {
 379       reinterpret_cast<ServerEnv*>(server_env_buffer),
 380       reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
 381   new (server_env[0]) ServerEnv;
 382   new (server_env[1]) ServerEnv;
 383   service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
 384                       &server_env[0]->response_writer, fixture->cq(),
 385                       fixture->cq(), tag(0));
 386   service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
 387                       &server_env[1]->response_writer, fixture->cq(),
 388                       fixture->cq(), tag(1));
 389   std::unique_ptr<EchoTestService::Stub> stub(
 390       EchoTestService::NewStub(fixture->channel()));
 391   auto inner_loop = [&](bool in_warmup) {
 392     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 393     recv_response.Clear();
 394     ClientContext cli_ctx;
 395     std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
 396         stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
 397     void* t;
 398     bool ok;
 399     response_reader->Finish(&recv_response, &recv_status, tag(4));
 400     TrickleCQNext(fixture.get(), &t, &ok, in_warmup ? -1 : state.iterations());
 401     GPR_ASSERT(ok);
 402     GPR_ASSERT(t == tag(0) || t == tag(1));
 403     intptr_t slot = reinterpret_cast<intptr_t>(t);
 404     ServerEnv* senv = server_env[slot];
 405     senv->response_writer.Finish(send_response, Status::OK, tag(3));
 406     for (int i = (1 << 3) | (1 << 4); i != 0;) {
 407       TrickleCQNext(fixture.get(), &t, &ok,
 408                     in_warmup ? -1 : state.iterations());
 409       GPR_ASSERT(ok);
 410       int tagnum = static_cast<int>(reinterpret_cast<intptr_t>(t));
 411       GPR_ASSERT(i & (1 << tagnum));
 412       i -= 1 << tagnum;
 413     }
 414     GPR_ASSERT(recv_status.ok());
 415
 416     senv->~ServerEnv();
 417     senv = new (senv) ServerEnv();
 418     service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
 419                         fixture->cq(), fixture->cq(), tag(slot));
 420   };
 421   gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
 422   for (int i = 0; i < GPR_MAX(absl::GetFlag(FLAGS_warmup_iterations),
 423                               absl::GetFlag(FLAGS_warmup_megabytes) * 1024 *
 424                                   1024 / (14 + state.range(0)));
 425        i++) {
 426     inner_loop(true);
 427     if (gpr_time_cmp(
 428             gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
 429             gpr_time_from_seconds(absl::GetFlag(FLAGS_warmup_max_time_seconds),
 430                                   GPR_TIMESPAN)) > 0) {
 431       break;
 432     }
 433   }
 434   while (state.KeepRunning()) {
 435     inner_loop(false);
 436   }
 437   fixture->Finish(state);
 438   fixture.reset();
 439   server_env[0]->~ServerEnv();
 440   server_env[1]->~ServerEnv();
 441   state.SetBytesProcessed(state.range(0) * state.iterations() +
 442                           state.range(1) * state.iterations());
 443 }
 444
 445 static void UnaryTrickleArgs(benchmark::internal::Benchmark* b) {
 446   for (int bw = 64; bw <= 128 * 1024 * 1024; bw *= 16) {
 447     b->Args({1, 1, bw});
 448     for (int i = 64; i <= 128 * 1024 * 1024; i *= 64) {
 449       double expected_time =
 450           static_cast<double>(14 + i) / (125.0 * static_cast<double>(bw));
 451       if (expected_time > 2.0) continue;
 452       b->Args({i, 1, bw});
 453       b->Args({1, i, bw});
 454       b->Args({i, i, bw});
 455     }
 456   }
 457 }
 458 BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
 459 }  // namespace testing
 460 }  // namespace grpc
 461
 462 extern gpr_timespec (*gpr_now_impl)(gpr_clock_type clock_type);
 463
 464 // Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
 465 // and others do not. This allows us to support both modes.
 466 namespace benchmark {
 467 void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
 468 }  // namespace benchmark
 469
 470 int main(int argc, char** argv) {
 471   grpc::testing::TestEnvironment env(argc, argv);
 472   LibraryInitializer libInit;
 473   ::benchmark::Initialize(&argc, argv);
 474   ::grpc::testing::InitTest(&argc, &argv, false);
 475   grpc_timer_manager_set_threading(false);
 476   gpr_now_impl = ::grpc::testing::fake_now;
 477   benchmark::RunTheBenchmarksNamespaced();
 478 }