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
  45 #include "src/cpp/client/create_channel_internal.h"
  46 #include "src/proto/grpc/testing/echo.grpc.pb.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   while (state.KeepRunning()) {
  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   while (state.KeepRunning()) {
 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   while (state.KeepRunning()) {
 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   while (state.KeepRunning()) {
 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   while (state.KeepRunning()) {
 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   grpc_channel* CreateChannel(const char* target,
 326                               const grpc_channel_args* args) override {
 327     return nullptr;
 328   }
 329 };
 330
 331 static grpc_arg StringArg(const char* key, const char* value) {
 332   grpc_arg a;
 333   a.type = GRPC_ARG_STRING;
 334   a.key = const_cast<char*>(key);
 335   a.value.string = const_cast<char*>(value);
 336   return a;
 337 }
 338
 339 enum FixtureFlags : uint32_t {
 340   CHECKS_NOT_LAST = 1,
 341   REQUIRES_TRANSPORT = 2,
 342 };
 343
 344 template <const grpc_channel_filter* kFilter, uint32_t kFlags>
 345 struct Fixture {
 346   const grpc_channel_filter* filter = kFilter;
 347   const uint32_t flags = kFlags;
 348 };
 349
 350 namespace dummy_filter {
 351
 352 static void StartTransportStreamOp(grpc_call_element* elem,
 353                                    grpc_transport_stream_op_batch* op) {}
 354
 355 static void StartTransportOp(grpc_channel_element* elem,
 356                              grpc_transport_op* op) {}
 357
 358 static grpc_error* InitCallElem(grpc_call_element* elem,
 359                                 const grpc_call_element_args* args) {
 360   return GRPC_ERROR_NONE;
 361 }
 362
 363 static void SetPollsetOrPollsetSet(grpc_call_element* elem,
 364                                    grpc_polling_entity* pollent) {}
 365
 366 static void DestroyCallElem(grpc_call_element* elem,
 367                             const grpc_call_final_info* final_info,
 368                             grpc_closure* then_sched_closure) {}
 369
 370 grpc_error* InitChannelElem(grpc_channel_element* elem,
 371                             grpc_channel_element_args* args) {
 372   return GRPC_ERROR_NONE;
 373 }
 374
 375 void DestroyChannelElem(grpc_channel_element* elem) {}
 376
 377 void GetChannelInfo(grpc_channel_element* elem,
 378                     const grpc_channel_info* channel_info) {}
 379
 380 static const grpc_channel_filter dummy_filter = {StartTransportStreamOp,
 381                                                  StartTransportOp,
 382                                                  0,
 383                                                  InitCallElem,
 384                                                  SetPollsetOrPollsetSet,
 385                                                  DestroyCallElem,
 386                                                  0,
 387                                                  InitChannelElem,
 388                                                  DestroyChannelElem,
 389                                                  GetChannelInfo,
 390                                                  "dummy_filter"};
 391
 392 }  // namespace dummy_filter
 393
 394 namespace dummy_transport {
 395
 396 /* Memory required for a single stream element - this is allocated by upper
 397    layers and initialized by the transport */
 398 size_t sizeof_stream; /* = sizeof(transport stream) */
 399
 400 /* name of this transport implementation */
 401 const char* name;
 402
 403 /* implementation of grpc_transport_init_stream */
 404 int InitStream(grpc_transport* self, grpc_stream* stream,
 405                grpc_stream_refcount* refcount, const void* server_data,
 406                grpc_core::Arena* arena) {
 407   return 0;
 408 }
 409
 410 /* implementation of grpc_transport_set_pollset */
 411 void SetPollset(grpc_transport* self, grpc_stream* stream,
 412                 grpc_pollset* pollset) {}
 413
 414 /* implementation of grpc_transport_set_pollset */
 415 void SetPollsetSet(grpc_transport* self, grpc_stream* stream,
 416                    grpc_pollset_set* pollset_set) {}
 417
 418 /* implementation of grpc_transport_perform_stream_op */
 419 void PerformStreamOp(grpc_transport* self, grpc_stream* stream,
 420                      grpc_transport_stream_op_batch* op) {
 421   GRPC_CLOSURE_SCHED(op->on_complete, GRPC_ERROR_NONE);
 422 }
 423
 424 /* implementation of grpc_transport_perform_op */
 425 void PerformOp(grpc_transport* self, grpc_transport_op* op) {}
 426
 427 /* implementation of grpc_transport_destroy_stream */
 428 void DestroyStream(grpc_transport* self, grpc_stream* stream,
 429                    grpc_closure* then_sched_closure) {}
 430
 431 /* implementation of grpc_transport_destroy */
 432 void Destroy(grpc_transport* self) {}
 433
 434 /* implementation of grpc_transport_get_endpoint */
 435 grpc_endpoint* GetEndpoint(grpc_transport* self) { return nullptr; }
 436
 437 static const grpc_transport_vtable dummy_transport_vtable = {
 438     0,          "dummy_http2", InitStream,
 439     SetPollset, SetPollsetSet, PerformStreamOp,
 440     PerformOp,  DestroyStream, Destroy,
 441     GetEndpoint};
 442
 443 static grpc_transport dummy_transport = {&dummy_transport_vtable};
 444
 445 }  // namespace dummy_transport
 446
 447 class NoOp {
 448  public:
 449   class Op {
 450    public:
 451     Op(NoOp* p, grpc_call_stack* s) {}
 452     void Finish() {}
 453   };
 454 };
 455
 456 class SendEmptyMetadata {
 457  public:
 458   SendEmptyMetadata() : op_payload_(nullptr) {
 459     memset(&op_, 0, sizeof(op_));
 460     op_.on_complete = GRPC_CLOSURE_INIT(&closure_, DoNothing, nullptr,
 461                                         grpc_schedule_on_exec_ctx);
 462     op_.send_initial_metadata = true;
 463     op_.payload = &op_payload_;
 464   }
 465
 466   class Op {
 467    public:
 468     Op(SendEmptyMetadata* p, grpc_call_stack* s) {
 469       grpc_metadata_batch_init(&batch_);
 470       p->op_payload_.send_initial_metadata.send_initial_metadata = &batch_;
 471     }
 472     void Finish() { grpc_metadata_batch_destroy(&batch_); }
 473
 474    private:
 475     grpc_metadata_batch batch_;
 476   };
 477
 478  private:
 479   const gpr_timespec deadline_ = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 480   const gpr_timespec start_time_ = gpr_now(GPR_CLOCK_MONOTONIC);
 481   const grpc_slice method_ = grpc_slice_from_static_string("/foo/bar");
 482   grpc_transport_stream_op_batch op_;
 483   grpc_transport_stream_op_batch_payload op_payload_;
 484   grpc_closure closure_;
 485 };
 486
 487 // Test a filter in isolation. Fixture specifies the filter under test (use the
 488 // Fixture<> template to specify this), and TestOp defines some unit of work to
 489 // perform on said filter.
 490 template <class Fixture, class TestOp>
 491 static void BM_IsolatedFilter(benchmark::State& state) {
 492   TrackCounters track_counters;
 493   Fixture fixture;
 494   std::ostringstream label;
 495   FakeClientChannelFactory fake_client_channel_factory;
 496
 497   std::vector<grpc_arg> args = {
 498       grpc_core::ClientChannelFactory::CreateChannelArg(
 499           &fake_client_channel_factory),
 500       StringArg(GRPC_ARG_SERVER_URI, "localhost"),
 501   };
 502   grpc_channel_args channel_args = {args.size(), &args[0]};
 503
 504   std::vector<const grpc_channel_filter*> filters;
 505   if (fixture.filter != nullptr) {
 506     filters.push_back(fixture.filter);
 507   }
 508   if (fixture.flags & CHECKS_NOT_LAST) {
 509     filters.push_back(&dummy_filter::dummy_filter);
 510     label << " #has_dummy_filter";
 511   }
 512
 513   grpc_core::ExecCtx exec_ctx;
 514   size_t channel_size = grpc_channel_stack_size(
 515       filters.size() == 0 ? nullptr : &filters[0], filters.size());
 516   grpc_channel_stack* channel_stack =
 517       static_cast<grpc_channel_stack*>(gpr_zalloc(channel_size));
 518   GPR_ASSERT(GRPC_LOG_IF_ERROR(
 519       "channel_stack_init",
 520       grpc_channel_stack_init(1, FilterDestroy, channel_stack, &filters[0],
 521                               filters.size(), &channel_args,
 522                               fixture.flags & REQUIRES_TRANSPORT
 523                                   ? &dummy_transport::dummy_transport
 524                                   : nullptr,
 525                               "CHANNEL", channel_stack)));
 526   grpc_core::ExecCtx::Get()->Flush();
 527   grpc_call_stack* call_stack =
 528       static_cast<grpc_call_stack*>(gpr_zalloc(channel_stack->call_stack_size));
 529   grpc_millis deadline = GRPC_MILLIS_INF_FUTURE;
 530   gpr_timespec start_time = gpr_now(GPR_CLOCK_MONOTONIC);
 531   grpc_slice method = grpc_slice_from_static_string("/foo/bar");
 532   grpc_call_final_info final_info;
 533   TestOp test_op_data;
 534   const int kArenaSize = 4096;
 535   grpc_call_element_args call_args{call_stack,
 536                                    nullptr,
 537                                    nullptr,
 538                                    method,
 539                                    start_time,
 540                                    deadline,
 541                                    grpc_core::Arena::Create(kArenaSize),
 542                                    nullptr};
 543   while (state.KeepRunning()) {
 544     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 545     GRPC_ERROR_UNREF(
 546         grpc_call_stack_init(channel_stack, 1, DoNothing, nullptr, &call_args));
 547     typename TestOp::Op op(&test_op_data, call_stack);
 548     grpc_call_stack_destroy(call_stack, &final_info, nullptr);
 549     op.Finish();
 550     grpc_core::ExecCtx::Get()->Flush();
 551     // recreate arena every 64k iterations to avoid oom
 552     if (0 == (state.iterations() & 0xffff)) {
 553       call_args.arena->Destroy();
 554       call_args.arena = grpc_core::Arena::Create(kArenaSize);
 555     }
 556   }
 557   call_args.arena->Destroy();
 558   grpc_channel_stack_destroy(channel_stack);
 559   grpc_core::ExecCtx::Get()->Flush();
 560
 561   gpr_free(channel_stack);
 562   gpr_free(call_stack);
 563
 564   state.SetLabel(label.str());
 565   track_counters.Finish(state);
 566 }
 567
 568 typedef Fixture<nullptr, 0> NoFilter;
 569 BENCHMARK_TEMPLATE(BM_IsolatedFilter, NoFilter, NoOp);
 570 typedef Fixture<&dummy_filter::dummy_filter, 0> DummyFilter;
 571 BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, NoOp);
 572 BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, SendEmptyMetadata);
 573 typedef Fixture<&grpc_client_channel_filter, 0> ClientChannelFilter;
 574 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientChannelFilter, NoOp);
 575 typedef Fixture<&grpc_message_compress_filter, CHECKS_NOT_LAST> CompressFilter;
 576 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, NoOp);
 577 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, SendEmptyMetadata);
 578 typedef Fixture<&grpc_client_deadline_filter, CHECKS_NOT_LAST>
 579     ClientDeadlineFilter;
 580 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientDeadlineFilter, NoOp);
 581 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientDeadlineFilter, SendEmptyMetadata);
 582 typedef Fixture<&grpc_server_deadline_filter, CHECKS_NOT_LAST>
 583     ServerDeadlineFilter;
 584 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ServerDeadlineFilter, NoOp);
 585 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ServerDeadlineFilter, SendEmptyMetadata);
 586 typedef Fixture<&grpc_http_client_filter, CHECKS_NOT_LAST | REQUIRES_TRANSPORT>
 587     HttpClientFilter;
 588 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpClientFilter, NoOp);
 589 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpClientFilter, SendEmptyMetadata);
 590 typedef Fixture<&grpc_http_server_filter, CHECKS_NOT_LAST> HttpServerFilter;
 591 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpServerFilter, NoOp);
 592 BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpServerFilter, SendEmptyMetadata);
 593 typedef Fixture<&grpc_message_size_filter, CHECKS_NOT_LAST> MessageSizeFilter;
 594 BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, NoOp);
 595 BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, SendEmptyMetadata);
 596 // This cmake target is disabled for now because it depends on OpenCensus, which
 597 // is Bazel-only.
 598 // typedef Fixture<&grpc_server_load_reporting_filter, CHECKS_NOT_LAST>
 599 //    LoadReportingFilter;
 600 // BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter, NoOp);
 601 // BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter,
 602 // SendEmptyMetadata);
 603
 604 ////////////////////////////////////////////////////////////////////////////////
 605 // Benchmarks isolating grpc_call
 606
 607 namespace isolated_call_filter {
 608
 609 typedef struct {
 610   grpc_core::CallCombiner* call_combiner;
 611 } call_data;
 612
 613 static void StartTransportStreamOp(grpc_call_element* elem,
 614                                    grpc_transport_stream_op_batch* op) {
 615   call_data* calld = static_cast<call_data*>(elem->call_data);
 616   // Construct list of closures to return.
 617   grpc_core::CallCombinerClosureList closures;
 618   if (op->recv_initial_metadata) {
 619     closures.Add(op->payload->recv_initial_metadata.recv_initial_metadata_ready,
 620                  GRPC_ERROR_NONE, "recv_initial_metadata");
 621   }
 622   if (op->recv_message) {
 623     closures.Add(op->payload->recv_message.recv_message_ready, GRPC_ERROR_NONE,
 624                  "recv_message");
 625   }
 626   if (op->recv_trailing_metadata) {
 627     closures.Add(
 628         op->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
 629         GRPC_ERROR_NONE, "recv_trailing_metadata");
 630   }
 631   if (op->on_complete != nullptr) {
 632     closures.Add(op->on_complete, GRPC_ERROR_NONE, "on_complete");
 633   }
 634   // Execute closures.
 635   closures.RunClosures(calld->call_combiner);
 636 }
 637
 638 static void StartTransportOp(grpc_channel_element* elem,
 639                              grpc_transport_op* op) {
 640   if (op->disconnect_with_error != GRPC_ERROR_NONE) {
 641     GRPC_ERROR_UNREF(op->disconnect_with_error);
 642   }
 643   GRPC_CLOSURE_SCHED(op->on_consumed, GRPC_ERROR_NONE);
 644 }
 645
 646 static grpc_error* InitCallElem(grpc_call_element* elem,
 647                                 const grpc_call_element_args* args) {
 648   call_data* calld = static_cast<call_data*>(elem->call_data);
 649   calld->call_combiner = args->call_combiner;
 650   return GRPC_ERROR_NONE;
 651 }
 652
 653 static void SetPollsetOrPollsetSet(grpc_call_element* elem,
 654                                    grpc_polling_entity* pollent) {}
 655
 656 static void DestroyCallElem(grpc_call_element* elem,
 657                             const grpc_call_final_info* final_info,
 658                             grpc_closure* then_sched_closure) {
 659   GRPC_CLOSURE_SCHED(then_sched_closure, GRPC_ERROR_NONE);
 660 }
 661
 662 grpc_error* InitChannelElem(grpc_channel_element* elem,
 663                             grpc_channel_element_args* args) {
 664   return GRPC_ERROR_NONE;
 665 }
 666
 667 void DestroyChannelElem(grpc_channel_element* elem) {}
 668
 669 void GetChannelInfo(grpc_channel_element* elem,
 670                     const grpc_channel_info* channel_info) {}
 671
 672 static const grpc_channel_filter isolated_call_filter = {
 673     StartTransportStreamOp,
 674     StartTransportOp,
 675     sizeof(call_data),
 676     InitCallElem,
 677     SetPollsetOrPollsetSet,
 678     DestroyCallElem,
 679     0,
 680     InitChannelElem,
 681     DestroyChannelElem,
 682     GetChannelInfo,
 683     "isolated_call_filter"};
 684 }  // namespace isolated_call_filter
 685
 686 class IsolatedCallFixture : public TrackCounters {
 687  public:
 688   IsolatedCallFixture() {
 689     grpc_channel_stack_builder* builder = grpc_channel_stack_builder_create();
 690     grpc_channel_stack_builder_set_name(builder, "dummy");
 691     grpc_channel_stack_builder_set_target(builder, "dummy_target");
 692     GPR_ASSERT(grpc_channel_stack_builder_append_filter(
 693         builder, &isolated_call_filter::isolated_call_filter, nullptr,
 694         nullptr));
 695     {
 696       grpc_core::ExecCtx exec_ctx;
 697       channel_ = grpc_channel_create_with_builder(builder, GRPC_CLIENT_CHANNEL);
 698     }
 699     cq_ = grpc_completion_queue_create_for_next(nullptr);
 700   }
 701
 702   void Finish(benchmark::State& state) {
 703     grpc_completion_queue_destroy(cq_);
 704     grpc_channel_destroy(channel_);
 705     TrackCounters::Finish(state);
 706   }
 707
 708   grpc_channel* channel() const { return channel_; }
 709   grpc_completion_queue* cq() const { return cq_; }
 710
 711  private:
 712   grpc_completion_queue* cq_;
 713   grpc_channel* channel_;
 714 };
 715
 716 static void BM_IsolatedCall_NoOp(benchmark::State& state) {
 717   IsolatedCallFixture fixture;
 718   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 719   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 720                                                 nullptr, nullptr);
 721   while (state.KeepRunning()) {
 722     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 723     grpc_call_unref(grpc_channel_create_registered_call(
 724         fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 725         method_hdl, deadline, nullptr));
 726   }
 727   fixture.Finish(state);
 728 }
 729 BENCHMARK(BM_IsolatedCall_NoOp);
 730
 731 static void BM_IsolatedCall_Unary(benchmark::State& state) {
 732   IsolatedCallFixture fixture;
 733   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 734   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 735                                                 nullptr, nullptr);
 736   grpc_slice slice = grpc_slice_from_static_string("hello world");
 737   grpc_byte_buffer* send_message = grpc_raw_byte_buffer_create(&slice, 1);
 738   grpc_byte_buffer* recv_message = nullptr;
 739   grpc_status_code status_code;
 740   grpc_slice status_details = grpc_empty_slice();
 741   grpc_metadata_array recv_initial_metadata;
 742   grpc_metadata_array_init(&recv_initial_metadata);
 743   grpc_metadata_array recv_trailing_metadata;
 744   grpc_metadata_array_init(&recv_trailing_metadata);
 745   grpc_op ops[6];
 746   memset(ops, 0, sizeof(ops));
 747   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
 748   ops[1].op = GRPC_OP_SEND_MESSAGE;
 749   ops[1].data.send_message.send_message = send_message;
 750   ops[2].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 751   ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;
 752   ops[3].data.recv_initial_metadata.recv_initial_metadata =
 753       &recv_initial_metadata;
 754   ops[4].op = GRPC_OP_RECV_MESSAGE;
 755   ops[4].data.recv_message.recv_message = &recv_message;
 756   ops[5].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 757   ops[5].data.recv_status_on_client.status = &status_code;
 758   ops[5].data.recv_status_on_client.status_details = &status_details;
 759   ops[5].data.recv_status_on_client.trailing_metadata = &recv_trailing_metadata;
 760   while (state.KeepRunning()) {
 761     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 762     grpc_call* call = grpc_channel_create_registered_call(
 763         fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 764         method_hdl, deadline, nullptr);
 765     grpc_call_start_batch(call, ops, 6, tag(1), nullptr);
 766     grpc_completion_queue_next(fixture.cq(),
 767                                gpr_inf_future(GPR_CLOCK_MONOTONIC), nullptr);
 768     grpc_call_unref(call);
 769   }
 770   fixture.Finish(state);
 771   grpc_metadata_array_destroy(&recv_initial_metadata);
 772   grpc_metadata_array_destroy(&recv_trailing_metadata);
 773   grpc_byte_buffer_destroy(send_message);
 774 }
 775 BENCHMARK(BM_IsolatedCall_Unary);
 776
 777 static void BM_IsolatedCall_StreamingSend(benchmark::State& state) {
 778   IsolatedCallFixture fixture;
 779   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
 780   void* method_hdl = grpc_channel_register_call(fixture.channel(), "/foo/bar",
 781                                                 nullptr, nullptr);
 782   grpc_slice slice = grpc_slice_from_static_string("hello world");
 783   grpc_byte_buffer* send_message = grpc_raw_byte_buffer_create(&slice, 1);
 784   grpc_metadata_array recv_initial_metadata;
 785   grpc_metadata_array_init(&recv_initial_metadata);
 786   grpc_metadata_array recv_trailing_metadata;
 787   grpc_metadata_array_init(&recv_trailing_metadata);
 788   grpc_op ops[2];
 789   memset(ops, 0, sizeof(ops));
 790   ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
 791   ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
 792   ops[1].data.recv_initial_metadata.recv_initial_metadata =
 793       &recv_initial_metadata;
 794   grpc_call* call = grpc_channel_create_registered_call(
 795       fixture.channel(), nullptr, GRPC_PROPAGATE_DEFAULTS, fixture.cq(),
 796       method_hdl, deadline, nullptr);
 797   grpc_call_start_batch(call, ops, 2, tag(1), nullptr);
 798   grpc_completion_queue_next(fixture.cq(), gpr_inf_future(GPR_CLOCK_MONOTONIC),
 799                              nullptr);
 800   memset(ops, 0, sizeof(ops));
 801   ops[0].op = GRPC_OP_SEND_MESSAGE;
 802   ops[0].data.send_message.send_message = send_message;
 803   while (state.KeepRunning()) {
 804     GPR_TIMER_SCOPE("BenchmarkCycle", 0);
 805     grpc_call_start_batch(call, ops, 1, tag(2), nullptr);
 806     grpc_completion_queue_next(fixture.cq(),
 807                                gpr_inf_future(GPR_CLOCK_MONOTONIC), nullptr);
 808   }
 809   grpc_call_unref(call);
 810   fixture.Finish(state);
 811   grpc_metadata_array_destroy(&recv_initial_metadata);
 812   grpc_metadata_array_destroy(&recv_trailing_metadata);
 813   grpc_byte_buffer_destroy(send_message);
 814 }
 815 BENCHMARK(BM_IsolatedCall_StreamingSend);
 816
 817 // Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
 818 // and others do not. This allows us to support both modes.
 819 namespace benchmark {
 820 void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
 821 }  // namespace benchmark
 822
 823 int main(int argc, char** argv) {
 824   LibraryInitializer libInit;
 825   ::benchmark::Initialize(&argc, argv);
 826   ::grpc::testing::InitTest(&argc, &argv, false);
 827   benchmark::RunTheBenchmarksNamespaced();
 828   return 0;
 829 }