test/cpp/microbenchmarks/bm_call_create.cc

   1 /*
   2  *
   3  * Copyright 2017 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 /* This benchmark exists to ensure that the benchmark integration is
  20  * working */
  21
  22 #include <benchmark/benchmark.h>
  23 #include <string.h>
  24 #include <sstream>
  25
  26 #include <grpc/grpc.h>
  27 #include <grpc/support/alloc.h>
  28 #include <grpc/support/string_util.h>
  29 #include <grpcpp/channel.h>
  30 #include <grpcpp/support/channel_arguments.h>
  31
  32 #include "src/core/ext/filters/client_channel/client_channel.h"
  33 #include "src/core/ext/filters/deadline/deadline_filter.h"
  34 #include "src/core/ext/filters/http/client/http_client_filter.h"
  35 #include "src/core/ext/filters/http/message_compress/message_compress_filter.h"
  36 #include "src/core/ext/filters/http/server/http_server_filter.h"
  37 #include "src/core/ext/filters/message_size/message_size_filter.h"
  38 #include "src/core/lib/channel/channel_stack.h"
  39 #include "src/core/lib/channel/connected_channel.h"
  40 #include "src/core/lib/iomgr/call_combiner.h"
  41 #include "src/core/lib/profiling/timers.h"
  42 #include "src/core/lib/surface/channel.h"
  43 #include "src/core/lib/transport/transport_impl.h"
  44 #include "src/cpp/client/create_channel_internal.h"
  45 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  46 #include "test/core/util/test_config.h"
  47 #include "test/cpp/microbenchmarks/helpers.h"
  48 #include "test/cpp/util/test_config.h"
  49
  50 void BM_Zalloc(benchmark::State& state) {
  51   // speed of light for call creation is zalloc, so benchmark a few interesting
  52   // sizes
  53   TrackCounters track_counters;
  54   size_t sz = state.range(0);
  55   for (auto _ : state) {
  56     gpr_free(gpr_zalloc(sz));
  57   }
  58   track_counters.Finish(state);
  59 }
  60 BENCHMARK(BM_Zalloc)
  61     ->Arg(64)
  62     ->Arg(128)
  63     ->Arg(256)
  64     ->Arg(512)
  65     ->Arg(1024)
  66     ->Arg(1536)
  67     ->Arg(2048)
  68     ->Arg(3072)
  69     ->Arg(4096)
  70     ->Arg(5120)
  71     ->Arg(6144)
  72     ->Arg(7168);
  73
  74 ////////////////////////////////////////////////////////////////////////////////
  75 // Benchmarks creating full stacks
  76
  77 class BaseChannelFixture {
  78  public:
  79   BaseChannelFixture(grpc_channel* channel) : channel_(channel) {}
  80   ~BaseChannelFixture() { grpc_channel_destroy(channel_); }
  81
  82   grpc_channel* channel() const { return channel_; }
  83
  84  private:
  85   grpc_channel* const channel_;
  86 };
  87
  88 class InsecureChannel : public BaseChannelFixture {
  89  public:
  90   InsecureChannel()
  91       : BaseChannelFixture(
  92             grpc_insecure_channel_create("localhost:1234", nullptr, nullptr)) {}
  93 };
  94
  95 class LameChannel : public BaseChannelFixture {
  96  public:
  97   LameChannel()
  98       : BaseChannelFixture(grpc_lame_client_channel_create(
  99             "localhost:1234", GRPC_STATUS_UNAUTHENTICATED, "blah")) {}
 100 };
 101
 102 template <class Fixture>
 103 static void BM_CallCreateDestroy(benchmark::State& state) {
 104   TrackCounters track_counters;
 105   Fixture fixture;
 106   grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
 107   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 108   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 109                                                 nullptr, nullptr);
 110   for (auto _ : state) {
 111     grpc_call_unref(grpc_channel_create_registered_call(
 112         fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, cq, method_hdl,
 113         deadline, nullptr));
 114   }
 115   grpc_completion_queue_destroy(cq);
 116   track_counters.Finish(state);
 117 }
 118
 119 BENCHMARK_TEMPLATE(BM_CallCreateDestroy, InsecureChannel);
 120 BENCHMARK_TEMPLATE(BM_CallCreateDestroy, LameChannel);
 121
 122 ////////////////////////////////////////////////////////////////////////////////
 123 // Benchmarks isolating individual filters
 124
 125 static void* tag(int i) {
 126   return reinterpret_cast<void*>(static_cast<intptr_t>(i));
 127 }
 128
 129 static void BM_LameChannelCallCreateCpp(benchmark::State& state) {
 130   TrackCounters track_counters;
 131   auto stub =
 132       grpc::testing::EchoTestService::NewStub(grpc::CreateChannelInternal(
 133           "",
 134           grpc_lame_client_channel_create("localhost:1234",
 135                                           GRPC_STATUS_UNAUTHENTICATED, "blah"),
 136           std::vector<std::unique_ptr<
 137               grpc::experimental::ClientInterceptorFactoryInterface>>()));
 138   grpc::CompletionQueue cq;
 139   grpc::testing::EchoRequest send_request;
 140   grpc::testing::EchoResponse recv_response;
 141   grpc::Status recv_status;
 142   for (auto _ : state) {
 143     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 144     grpc::ClientContext cli_ctx;
 145     auto reader = stub->AsyncEcho(&cli_ctx, send_request, &cq);
 146     reader->Finish(&recv_response, &recv_status, tag(0));
 147     void* t;
 148     bool ok;
 149     GPR_ASSERT(cq.Next(&t, &ok));
 150     GPR_ASSERT(ok);
 151   }
 152   track_counters.Finish(state);
 153 }
 154 BENCHMARK(BM_LameChannelCallCreateCpp);
 155
 156 static void do_nothing(void* /*ignored*/) {}
 157
 158 static void BM_LameChannelCallCreateCore(benchmark::State& state) {
 159   TrackCounters track_counters;
 160
 161   grpc_channel* channel;
 162   grpc_completion_queue* cq;
 163   grpc_metadata_array initial_metadata_recv;
 164   grpc_metadata_array trailing_metadata_recv;
 165   grpc_byte_buffer* response_payload_recv = nullptr;
 166   grpc_status_code status;
 167   grpc_slice details;
 168   grpc::testing::EchoRequest send_request;
 169   grpc_slice send_request_slice =
 170       grpc_slice_new(&send_request, sizeof(send_request), do_nothing);
 171
 172   channel = grpc_lame_client_channel_create(
 173       "localhost:1234", GRPC_STATUS_UNAUTHENTICATED, "blah");
 174   cq = grpc_completion_queue_create_for_next(nullptr);
 175   void* rc = grpc_channel_register_call(
 176       channel, "/grpc.testing.EchoTestService/Echo", nullptr, nullptr);
 177   for (auto _ : state) {
 178     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 179     grpc_call* call = grpc_channel_create_registered_call(
 180         channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq, rc,
 181         gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
 182     grpc_metadata_array_init(&initial_metadata_recv);
 183     grpc_metadata_array_init(&trailing_metadata_recv);
 184     grpc_byte_buffer* request_payload_send =
 185         grpc_raw_byte_buffer_create(&send_request_slice, 1);
 186
 187     // Fill in call ops
 188     grpc_op ops[6];
 189     memset(ops, 0, sizeof(ops));
 190     grpc_op* op = ops;
 191     op->op = GRPC_OP_SEND_INITIAL_METADATA;
 192     op->data.send_initial_metadata.count = 0;
 193     op++;
 194     op->op = GRPC_OP_SEND_MESSAGE;
 195     op->data.send_message.send_message = request_payload_send;
 196     op++;
 197     op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 198     op++;
 199     op->op = GRPC_OP_RECV_INITIAL_METADATA;
 200     op->data.recv_initial_metadata.recv_initial_metadata =
 201         &initial_metadata_recv;
 202     op++;
 203     op->op = GRPC_OP_RECV_MESSAGE;
 204     op->data.recv_message.recv_message = &response_payload_recv;
 205     op++;
 206     op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 207     op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 208     op->data.recv_status_on_client.status = &status;
 209     op->data.recv_status_on_client.status_details = &details;
 210     op++;
 211
 212     GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, ops,
 213                                                      (size_t)(op - ops),
 214                                                      (void*)1, nullptr));
 215     grpc_event ev = grpc_completion_queue_next(
 216         cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
 217     GPR_ASSERT(ev.type != GRPC_QUEUE_SHUTDOWN);
 218     GPR_ASSERT(ev.success != 0);
 219     grpc_call_unref(call);
 220     grpc_byte_buffer_destroy(request_payload_send);
 221     grpc_byte_buffer_destroy(response_payload_recv);
 222     grpc_metadata_array_destroy(&initial_metadata_recv);
 223     grpc_metadata_array_destroy(&trailing_metadata_recv);
 224   }
 225   grpc_channel_destroy(channel);
 226   grpc_completion_queue_destroy(cq);
 227   grpc_slice_unref(send_request_slice);
 228   track_counters.Finish(state);
 229 }
 230 BENCHMARK(BM_LameChannelCallCreateCore);
 231
 232 static void BM_LameChannelCallCreateCoreSeparateBatch(benchmark::State& state) {
 233   TrackCounters track_counters;
 234
 235   grpc_channel* channel;
 236   grpc_completion_queue* cq;
 237   grpc_metadata_array initial_metadata_recv;
 238   grpc_metadata_array trailing_metadata_recv;
 239   grpc_byte_buffer* response_payload_recv = nullptr;
 240   grpc_status_code status;
 241   grpc_slice details;
 242   grpc::testing::EchoRequest send_request;
 243   grpc_slice send_request_slice =
 244       grpc_slice_new(&send_request, sizeof(send_request), do_nothing);
 245
 246   channel = grpc_lame_client_channel_create(
 247       "localhost:1234", GRPC_STATUS_UNAUTHENTICATED, "blah");
 248   cq = grpc_completion_queue_create_for_next(nullptr);
 249   void* rc = grpc_channel_register_call(
 250       channel, "/grpc.testing.EchoTestService/Echo", nullptr, nullptr);
 251   for (auto _ : state) {
 252     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 253     grpc_call* call = grpc_channel_create_registered_call(
 254         channel, nullptr, GRPC_PROPAGATE_DEFAULTS, cq, rc,
 255         gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
 256     grpc_metadata_array_init(&initial_metadata_recv);
 257     grpc_metadata_array_init(&trailing_metadata_recv);
 258     grpc_byte_buffer* request_payload_send =
 259         grpc_raw_byte_buffer_create(&send_request_slice, 1);
 260
 261     // Fill in call ops
 262     grpc_op ops[3];
 263     memset(ops, 0, sizeof(ops));
 264     grpc_op* op = ops;
 265     op->op = GRPC_OP_SEND_INITIAL_METADATA;
 266     op->data.send_initial_metadata.count = 0;
 267     op++;
 268     op->op = GRPC_OP_SEND_MESSAGE;
 269     op->data.send_message.send_message = request_payload_send;
 270     op++;
 271     op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 272     op++;
 273     GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, ops,
 274                                                      (size_t)(op - ops),
 275                                                      (void*)nullptr, nullptr));
 276     memset(ops, 0, sizeof(ops));
 277     op = ops;
 278     op->op = GRPC_OP_RECV_INITIAL_METADATA;
 279     op->data.recv_initial_metadata.recv_initial_metadata =
 280         &initial_metadata_recv;
 281     op++;
 282     op->op = GRPC_OP_RECV_MESSAGE;
 283     op->data.recv_message.recv_message = &response_payload_recv;
 284     op++;
 285     op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 286     op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 287     op->data.recv_status_on_client.status = &status;
 288     op->data.recv_status_on_client.status_details = &details;
 289     op++;
 290
 291     GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, ops,
 292                                                      (size_t)(op - ops),
 293                                                      (void*)1, nullptr));
 294     grpc_event ev = grpc_completion_queue_next(
 295         cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
 296     GPR_ASSERT(ev.type != GRPC_QUEUE_SHUTDOWN);
 297     GPR_ASSERT(ev.success == 0);
 298     ev = grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
 299                                     nullptr);
 300     GPR_ASSERT(ev.type != GRPC_QUEUE_SHUTDOWN);
 301     GPR_ASSERT(ev.success != 0);
 302     grpc_call_unref(call);
 303     grpc_byte_buffer_destroy(request_payload_send);
 304     grpc_byte_buffer_destroy(response_payload_recv);
 305     grpc_metadata_array_destroy(&initial_metadata_recv);
 306     grpc_metadata_array_destroy(&trailing_metadata_recv);
 307   }
 308   grpc_channel_destroy(channel);
 309   grpc_completion_queue_destroy(cq);
 310   grpc_slice_unref(send_request_slice);
 311   track_counters.Finish(state);
 312 }
 313 BENCHMARK(BM_LameChannelCallCreateCoreSeparateBatch);
 314
 315 static void FilterDestroy(void* arg, grpc_error* /*error*/) { gpr_free(arg); }
 316
 317 static void DoNothing(void* /*arg*/, grpc_error* /*error*/) {}
 318
 319 class FakeClientChannelFactory : public grpc_core::ClientChannelFactory {
 320  public:
 321   grpc_core::Subchannel* CreateSubchannel(
 322       const grpc_channel_args* /*args*/) override {
 323     return nullptr;
 324   }
 325 };
 326
 327 static grpc_arg StringArg(const char* key, const char* value) {
 328   grpc_arg a;
 329   a.type = GRPC_ARG_STRING;
 330   a.key = const_cast<char*>(key);
 331   a.value.string = const_cast<char*>(value);
 332   return a;
 333 }
 334
 335 enum FixtureFlags : uint32_t {
 336   CHECKS_NOT_LAST = 1,
 337   REQUIRES_TRANSPORT = 2,
 338 };
 339
 340 template <const grpc_channel_filter* kFilter, uint32_t kFlags>
 341 struct Fixture {
 342   const grpc_channel_filter* filter = kFilter;
 343   const uint32_t flags = kFlags;
 344 };
 345
 346 namespace dummy_filter {
 347
 348 static void StartTransportStreamOp(grpc_call_element* /*elem*/,
 349                                    grpc_transport_stream_op_batch* /*op*/) {}
 350
 351 static void StartTransportOp(grpc_channel_element* /*elem*/,
 352                              grpc_transport_op* /*op*/) {}
 353
 354 static grpc_error* InitCallElem(grpc_call_element* /*elem*/,
 355                                 const grpc_call_element_args* /*args*/) {
 356   return GRPC_ERROR_NONE;
 357 }
 358
 359 static void SetPollsetOrPollsetSet(grpc_call_element* /*elem*/,
 360                                    grpc_polling_entity* /*pollent*/) {}
 361
 362 static void DestroyCallElem(grpc_call_element* /*elem*/,
 363                             const grpc_call_final_info* /*final_info*/,
 364                             grpc_closure* /*then_sched_closure*/) {}
 365
 366 grpc_error* InitChannelElem(grpc_channel_element* /*elem*/,
 367                             grpc_channel_element_args* /*args*/) {
 368   return GRPC_ERROR_NONE;
 369 }
 370
 371 void DestroyChannelElem(grpc_channel_element* /*elem*/) {}
 372
 373 void GetChannelInfo(grpc_channel_element* /*elem*/,
 374                     const grpc_channel_info* /*channel_info*/) {}
 375
 376 static const grpc_channel_filter dummy_filter = {StartTransportStreamOp,
 377                                                  StartTransportOp,
 378                                                  0,
 379                                                  InitCallElem,
 380                                                  SetPollsetOrPollsetSet,
 381                                                  DestroyCallElem,
 382                                                  0,
 383                                                  InitChannelElem,
 384                                                  DestroyChannelElem,
 385                                                  GetChannelInfo,
 386                                                  "dummy_filter"};
 387
 388 }  // namespace dummy_filter
 389
 390 namespace dummy_transport {
 391
 392 /* Memory required for a single stream element - this is allocated by upper
 393    layers and initialized by the transport */
 394 size_t sizeof_stream; /* = sizeof(transport stream) */
 395
 396 /* name of this transport implementation */
 397 const char* name;
 398
 399 /* implementation of grpc_transport_init_stream */
 400 int InitStream(grpc_transport* /*self*/, grpc_stream* /*stream*/,
 401                grpc_stream_refcount* /*refcount*/, const void* /*server_data*/,
 402                grpc_core::Arena* /*arena*/) {
 403   return 0;
 404 }
 405
 406 /* implementation of grpc_transport_set_pollset */
 407 void SetPollset(grpc_transport* /*self*/, grpc_stream* /*stream*/,
 408                 grpc_pollset* /*pollset*/) {}
 409
 410 /* implementation of grpc_transport_set_pollset */
 411 void SetPollsetSet(grpc_transport* /*self*/, grpc_stream* /*stream*/,
 412                    grpc_pollset_set* /*pollset_set*/) {}
 413
 414 /* implementation of grpc_transport_perform_stream_op */
 415 void PerformStreamOp(grpc_transport* /*self*/, grpc_stream* /*stream*/,
 416                      grpc_transport_stream_op_batch* op) {
 417   grpc_core::ExecCtx::Run(DEBUG_LOCATION, op->on_complete, GRPC_ERROR_NONE);
 418 }
 419
 420 /* implementation of grpc_transport_perform_op */
 421 void PerformOp(grpc_transport* /*self*/, grpc_transport_op* /*op*/) {}
 422
 423 /* implementation of grpc_transport_destroy_stream */
 424 void DestroyStream(grpc_transport* /*self*/, grpc_stream* /*stream*/,
 425                    grpc_closure* /*then_sched_closure*/) {}
 426
 427 /* implementation of grpc_transport_destroy */
 428 void Destroy(grpc_transport* /*self*/) {}
 429
 430 /* implementation of grpc_transport_get_endpoint */
 431 grpc_endpoint* GetEndpoint(grpc_transport* /*self*/) { return nullptr; }
 432
 433 static const grpc_transport_vtable dummy_transport_vtable = {
 434     0,          "dummy_http2", InitStream,
 435     SetPollset, SetPollsetSet, PerformStreamOp,
 436     PerformOp,  DestroyStream, Destroy,
 437     GetEndpoint};
 438
 439 static grpc_transport dummy_transport = {&dummy_transport_vtable};
 440
 441 }  // namespace dummy_transport
 442
 443 class NoOp {
 444  public:
 445   class Op {
 446    public:
 447     Op(NoOp* /*p*/, grpc_call_stack* /*s*/) {}
 448     void Finish() {}
 449   };
 450 };
 451
 452 class SendEmptyMetadata {
 453  public:
 454   SendEmptyMetadata() : op_payload_(nullptr) {
 455     op_ = {};
 456     op_.on_complete = GRPC_CLOSURE_INIT(&closure_, DoNothing, nullptr,
 457                                         grpc_schedule_on_exec_ctx);
 458     op_.send_initial_metadata = true;
 459     op_.payload = &op_payload_;
 460   }
 461
 462   class Op {
 463    public:
 464     Op(SendEmptyMetadata* p, grpc_call_stack* /*s*/) {
 465       grpc_metadata_batch_init(&batch_);
 466       p->op_payload_.send_initial_metadata.send_initial_metadata = &batch_;
 467     }
 468     void Finish() { grpc_metadata_batch_destroy(&batch_); }
 469
 470    private:
 471     grpc_metadata_batch batch_;
 472   };
 473
 474  private:
 475   const gpr_timespec deadline_ = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 476   const gpr_timespec start_time_ = gpr_now(GPR_CLOCK_MONOTONIC);
 477   const grpc_slice method_ = grpc_slice_from_static_string("/foo/bar");
 478   grpc_transport_stream_op_batch op_;
 479   grpc_transport_stream_op_batch_payload op_payload_;
 480   grpc_closure closure_;
 481 };
 482
 483 // Test a filter in isolation. Fixture specifies the filter under test (use the
 484 // Fixture<> template to specify this), and TestOp defines some unit of work to
 485 // perform on said filter.
 486 template <class Fixture, class TestOp>
 487 static void BM_IsolatedFilter(benchmark::State& state) {
 488   TrackCounters track_counters;
 489   Fixture fixture;
 490   std::ostringstream label;
 491   FakeClientChannelFactory fake_client_channel_factory;
 492
 493   std::vector<grpc_arg> args = {
 494       grpc_core::ClientChannelFactory::CreateChannelArg(
 495           &fake_client_channel_factory),
 496       StringArg(GRPC_ARG_SERVER_URI, "localhost"),
 497   };
 498   grpc_channel_args channel_args = {args.size(), &args[0]};
 499
 500   std::vector<const grpc_channel_filter*> filters;
 501   if (fixture.filter != nullptr) {
 502     filters.push_back(fixture.filter);
 503   }
 504   if (fixture.flags & CHECKS_NOT_LAST) {
 505     filters.push_back(&dummy_filter::dummy_filter);
 506     label << " #has_dummy_filter";
 507   }
 508
 509   grpc_core::ExecCtx exec_ctx;
 510   size_t channel_size = grpc_channel_stack_size(
 511       filters.empty() ? nullptr : &filters[0], filters.size());
 512   grpc_channel_stack* channel_stack =
 513       static_cast<grpc_channel_stack*>(gpr_zalloc(channel_size));
 514   GPR_ASSERT(GRPC_LOG_IF_ERROR(
 515       "channel_stack_init",
 516       grpc_channel_stack_init(1, FilterDestroy, channel_stack,
 517                               filters.empty() ? nullptr : &filters[0],
 518                               filters.size(), &channel_args,
 519                               fixture.flags & REQUIRES_TRANSPORT
 520                                   ? &dummy_transport::dummy_transport
 521                                   : nullptr,
 522                               "CHANNEL", channel_stack)));
 523   grpc_core::ExecCtx::Get()->Flush();
 524   grpc_call_stack* call_stack =
 525       static_cast<grpc_call_stack*>(gpr_zalloc(channel_stack->call_stack_size));
 526   grpc_millis deadline = GRPC_MILLIS_INF_FUTURE;
 527   gpr_cycle_counter start_time = gpr_get_cycle_counter();
 528   grpc_slice method = grpc_slice_from_static_string("/foo/bar");
 529   grpc_call_final_info final_info;
 530   TestOp test_op_data;
 531   const int kArenaSize = 4096;
 532   grpc_call_context_element context[GRPC_CONTEXT_COUNT] = {};
 533   grpc_call_element_args call_args{call_stack,
 534                                    nullptr,
 535                                    context,
 536                                    method,
 537                                    start_time,
 538                                    deadline,
 539                                    grpc_core::Arena::Create(kArenaSize),
 540                                    nullptr};
 541   while (state.KeepRunning()) {
 542     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 543     GRPC_ERROR_UNREF(
 544         grpc_call_stack_init(channel_stack, 1, DoNothing, nullptr, &call_args));
 545     typename TestOp::Op op(&test_op_data, call_stack);
 546     grpc_call_stack_destroy(call_stack, &final_info, nullptr);
 547     op.Finish();
 548     grpc_core::ExecCtx::Get()->Flush();
 549     // recreate arena every 64k iterations to avoid oom
 550     if (0 == (state.iterations() & 0xffff)) {
 551       call_args.arena->Destroy();
 552       call_args.arena = grpc_core::Arena::Create(kArenaSize);
 553     }
 554   }
 555   call_args.arena->Destroy();
 556   grpc_channel_stack_destroy(channel_stack);
 557   grpc_core::ExecCtx::Get()->Flush();
 558
 559   gpr_free(channel_stack);
 560   gpr_free(call_stack);
 561
 562   state.SetLabel(label.str());
 563   track_counters.Finish(state);
 564 }
 565
 566 typedef Fixture<nullptr, 0> NoFilter;
 567 BENCHMARK_TEMPLATE(BM_IsolatedFilter, NoFilter, NoOp);
 568 typedef Fixture<&dummy_filter::dummy_filter, 0> DummyFilter;
 569 BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, NoOp);
 570 BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, SendEmptyMetadata);
 571 typedef Fixture<&grpc_client_channel_filter, 0> ClientChannelFilter;
 572 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientChannelFilter, NoOp);
 573 typedef Fixture<&grpc_message_compress_filter, CHECKS_NOT_LAST> CompressFilter;
 574 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, NoOp);
 575 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, SendEmptyMetadata);
 576 typedef Fixture<&grpc_client_deadline_filter, CHECKS_NOT_LAST>
 577     ClientDeadlineFilter;
 578 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientDeadlineFilter, NoOp);
 579 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientDeadlineFilter, SendEmptyMetadata);
 580 typedef Fixture<&grpc_server_deadline_filter, CHECKS_NOT_LAST>
 581     ServerDeadlineFilter;
 582 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ServerDeadlineFilter, NoOp);
 583 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ServerDeadlineFilter, SendEmptyMetadata);
 584 typedef Fixture<&grpc_http_client_filter, CHECKS_NOT_LAST | REQUIRES_TRANSPORT>
 585     HttpClientFilter;
 586 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpClientFilter, NoOp);
 587 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpClientFilter, SendEmptyMetadata);
 588 typedef Fixture<&grpc_http_server_filter, CHECKS_NOT_LAST> HttpServerFilter;
 589 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpServerFilter, NoOp);
 590 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpServerFilter, SendEmptyMetadata);
 591 typedef Fixture<&grpc_message_size_filter, CHECKS_NOT_LAST> MessageSizeFilter;
 592 BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, NoOp);
 593 BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, SendEmptyMetadata);
 594 // This cmake target is disabled for now because it depends on OpenCensus, which
 595 // is Bazel-only.
 596 // typedef Fixture<&grpc_server_load_reporting_filter, CHECKS_NOT_LAST>
 597 //    LoadReportingFilter;
 598 // BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter, NoOp);
 599 // BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter,
 600 // SendEmptyMetadata);
 601
 602 ////////////////////////////////////////////////////////////////////////////////
 603 // Benchmarks isolating grpc_call
 604
 605 namespace isolated_call_filter {
 606
 607 typedef struct {
 608   grpc_core::CallCombiner* call_combiner;
 609 } call_data;
 610
 611 static void StartTransportStreamOp(grpc_call_element* elem,
 612                                    grpc_transport_stream_op_batch* op) {
 613   call_data* calld = static_cast<call_data*>(elem->call_data);
 614   // Construct list of closures to return.
 615   grpc_core::CallCombinerClosureList closures;
 616   if (op->recv_initial_metadata) {
 617     closures.Add(op->payload->recv_initial_metadata.recv_initial_metadata_ready,
 618                  GRPC_ERROR_NONE, "recv_initial_metadata");
 619   }
 620   if (op->recv_message) {
 621     closures.Add(op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE,
 622                  "recv_message");
 623   }
 624   if (op->recv_trailing_metadata) {
 625     closures.Add(
 626         op->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
 627         GRPC_ERROR_NONE, "recv_trailing_metadata");
 628   }
 629   if (op->on_complete != nullptr) {
 630     closures.Add(op->on_complete, GRPC_ERROR_NONE, "on_complete");
 631   }
 632   // Execute closures.
 633   closures.RunClosures(calld->call_combiner);
 634 }
 635
 636 static void StartTransportOp(grpc_channel_element* /*elem*/,
 637                              grpc_transport_op* op) {
 638   if (op->disconnect_with_error != GRPC_ERROR_NONE) {
 639     GRPC_ERROR_UNREF(op->disconnect_with_error);
 640   }
 641   grpc_core::ExecCtx::Run(DEBUG_LOCATION, op->on_consumed, GRPC_ERROR_NONE);
 642 }
 643
 644 static grpc_error* InitCallElem(grpc_call_element* elem,
 645                                 const grpc_call_element_args* args) {
 646   call_data* calld = static_cast<call_data*>(elem->call_data);
 647   calld->call_combiner = args->call_combiner;
 648   return GRPC_ERROR_NONE;
 649 }
 650
 651 static void SetPollsetOrPollsetSet(grpc_call_element* /*elem*/,
 652                                    grpc_polling_entity* /*pollent*/) {}
 653
 654 static void DestroyCallElem(grpc_call_element* /*elem*/,
 655                             const grpc_call_final_info* /*final_info*/,
 656                             grpc_closure* then_sched_closure) {
 657   grpc_core::ExecCtx::Run(DEBUG_LOCATION, then_sched_closure, GRPC_ERROR_NONE);
 658 }
 659
 660 grpc_error* InitChannelElem(grpc_channel_element* /*elem*/,
 661                             grpc_channel_element_args* /*args*/) {
 662   return GRPC_ERROR_NONE;
 663 }
 664
 665 void DestroyChannelElem(grpc_channel_element* /*elem*/) {}
 666
 667 void GetChannelInfo(grpc_channel_element* /*elem*/,
 668                     const grpc_channel_info* /*channel_info*/) {}
 669
 670 static const grpc_channel_filter isolated_call_filter = {
 671     StartTransportStreamOp,
 672     StartTransportOp,
 673     sizeof(call_data),
 674     InitCallElem,
 675     SetPollsetOrPollsetSet,
 676     DestroyCallElem,
 677     0,
 678     InitChannelElem,
 679     DestroyChannelElem,
 680     GetChannelInfo,
 681     "isolated_call_filter"};
 682 }  // namespace isolated_call_filter
 683
 684 class IsolatedCallFixture : public TrackCounters {
 685  public:
 686   IsolatedCallFixture() {
 687     // We are calling grpc_channel_stack_builder_create() instead of
 688     // grpc_channel_create() here, which means we're not getting the
 689     // grpc_init() called by grpc_channel_create(), but we are getting
 690     // the grpc_shutdown() run by grpc_channel_destroy().  So we need to
 691     // call grpc_init() manually here to balance things out.
 692     grpc_init();
 693     grpc_channel_stack_builder* builder = grpc_channel_stack_builder_create();
 694     grpc_channel_stack_builder_set_name(builder, "dummy");
 695     grpc_channel_stack_builder_set_target(builder, "dummy_target");
 696     GPR_ASSERT(grpc_channel_stack_builder_append_filter(
 697         builder, &isolated_call_filter::isolated_call_filter, nullptr,
 698         nullptr));
 699     {
 700       grpc_core::ExecCtx exec_ctx;
 701       channel_ = grpc_channel_create_with_builder(builder, GRPC_CLIENT_CHANNEL);
 702     }
 703     cq_ = grpc_completion_queue_create_for_next(nullptr);
 704   }
 705
 706   void Finish(benchmark::State& state) override {
 707     grpc_completion_queue_destroy(cq_);
 708     grpc_channel_destroy(channel_);
 709     TrackCounters::Finish(state);
 710   }
 711
 712   grpc_channel* channel() const { return channel_; }
 713   grpc_completion_queue* cq() const { return cq_; }
 714
 715  private:
 716   grpc_completion_queue* cq_;
 717   grpc_channel* channel_;
 718 };
 719
 720 static void BM_IsolatedCall_NoOp(benchmark::State& state) {
 721   IsolatedCallFixture fixture;
 722   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 723   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 724                                                 nullptr, nullptr);
 725   for (auto _ : state) {
 726     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 727     grpc_call_unref(grpc_channel_create_registered_call(
 728         fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 729         method_hdl, deadline, nullptr));
 730   }
 731   fixture.Finish(state);
 732 }
 733 BENCHMARK(BM_IsolatedCall_NoOp);
 734
 735 static void BM_IsolatedCall_Unary(benchmark::State& state) {
 736   IsolatedCallFixture fixture;
 737   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 738   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 739                                                 nullptr, nullptr);
 740   grpc_slice slice = grpc_slice_from_static_string("hello world");
 741   grpc_byte_buffer* send_message = grpc_raw_byte_buffer_create(&slice, 1);
 742   grpc_byte_buffer* recv_message = nullptr;
 743   grpc_status_code status_code;
 744   grpc_slice status_details = grpc_empty_slice();
 745   grpc_metadata_array recv_initial_metadata;
 746   grpc_metadata_array_init(&recv_initial_metadata);
 747   grpc_metadata_array recv_trailing_metadata;
 748   grpc_metadata_array_init(&recv_trailing_metadata);
 749   grpc_op ops[6];
 750   memset(ops, 0, sizeof(ops));
 751   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
 752   ops[1].op = GRPC_OP_SEND_MESSAGE;
 753   ops[1].data.send_message.send_message = send_message;
 754   ops[2].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 755   ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
 756   ops[3].data.recv_initial_metadata.recv_initial_metadata =
 757       &recv_initial_metadata;
 758   ops[4].op = GRPC_OP_RECV_MESSAGE;
 759   ops[4].data.recv_message.recv_message = &recv_message;
 760   ops[5].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 761   ops[5].data.recv_status_on_client.status = &status_code;
 762   ops[5].data.recv_status_on_client.status_details = &status_details;
 763   ops[5].data.recv_status_on_client.trailing_metadata = &recv_trailing_metadata;
 764   for (auto _ : state) {
 765     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 766     grpc_call* call = grpc_channel_create_registered_call(
 767         fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 768         method_hdl, deadline, nullptr);
 769     grpc_call_start_batch(call, ops, 6, tag(1), nullptr);
 770     grpc_completion_queue_next(fixture.cq(),
 771                                gpr_inf_future(GPR_CLOCK_MONOTONIC), nullptr);
 772     grpc_call_unref(call);
 773   }
 774   fixture.Finish(state);
 775   grpc_metadata_array_destroy(&recv_initial_metadata);
 776   grpc_metadata_array_destroy(&recv_trailing_metadata);
 777   grpc_byte_buffer_destroy(send_message);
 778 }
 779 BENCHMARK(BM_IsolatedCall_Unary);
 780
 781 static void BM_IsolatedCall_StreamingSend(benchmark::State& state) {
 782   IsolatedCallFixture fixture;
 783   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 784   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 785                                                 nullptr, nullptr);
 786   grpc_slice slice = grpc_slice_from_static_string("hello world");
 787   grpc_byte_buffer* send_message = grpc_raw_byte_buffer_create(&slice, 1);
 788   grpc_metadata_array recv_initial_metadata;
 789   grpc_metadata_array_init(&recv_initial_metadata);
 790   grpc_metadata_array recv_trailing_metadata;
 791   grpc_metadata_array_init(&recv_trailing_metadata);
 792   grpc_op ops[2];
 793   memset(ops, 0, sizeof(ops));
 794   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
 795   ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
 796   ops[1].data.recv_initial_metadata.recv_initial_metadata =
 797       &recv_initial_metadata;
 798   grpc_call* call = grpc_channel_create_registered_call(
 799       fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 800       method_hdl, deadline, nullptr);
 801   grpc_call_start_batch(call, ops, 2, tag(1), nullptr);
 802   grpc_completion_queue_next(fixture.cq(), gpr_inf_future(GPR_CLOCK_MONOTONIC),
 803                              nullptr);
 804   memset(ops, 0, sizeof(ops));
 805   ops[0].op = GRPC_OP_SEND_MESSAGE;
 806   ops[0].data.send_message.send_message = send_message;
 807   for (auto _ : state) {
 808     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 809     grpc_call_start_batch(call, ops, 1, tag(2), nullptr);
 810     grpc_completion_queue_next(fixture.cq(),
 811                                gpr_inf_future(GPR_CLOCK_MONOTONIC), nullptr);
 812   }
 813   grpc_call_unref(call);
 814   fixture.Finish(state);
 815   grpc_metadata_array_destroy(&recv_initial_metadata);
 816   grpc_metadata_array_destroy(&recv_trailing_metadata);
 817   grpc_byte_buffer_destroy(send_message);
 818 }
 819 BENCHMARK(BM_IsolatedCall_StreamingSend);
 820
 821 // Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
 822 // and others do not. This allows us to support both modes.
 823 namespace benchmark {
 824 void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
 825 }  // namespace benchmark
 826
 827 int main(int argc, char** argv) {
 828   grpc::testing::TestEnvironment env(argc, argv);
 829   LibraryInitializer libInit;
 830   ::benchmark::Initialize(&argc, argv);
 831   ::grpc::testing::InitTest(&argc, &argv, false);
 832   benchmark::RunTheBenchmarksNamespaced();
 833   return 0;
 834 }