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