test/cpp/microbenchmarks/bm_chttp2_transport.cc - third_party/grpc - Git at Google

 /*
  *
  * Copyright 2015 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  *
  */

 /* Microbenchmarks around CHTTP2 transport operations */

 #include <grpc++/support/channel_arguments.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <string.h>
 #include <memory>
 #include <queue>
 #include <sstream>
 extern "C" {
 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/ext/transport/chttp2/transport/internal.h"
 #include "src/core/lib/iomgr/closure.h"
 #include "src/core/lib/iomgr/resource_quota.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/transport/static_metadata.h"
 }
 #include "test/cpp/microbenchmarks/helpers.h"
 #include "third_party/benchmark/include/benchmark/benchmark.h"

 auto &force_library_initialization = Library::get();

 ////////////////////////////////////////////////////////////////////////////////
 // Helper classes
 //

 class DummyEndpoint : public grpc_endpoint {
  public:
   DummyEndpoint() {
     static const grpc_endpoint_vtable my_vtable = {
         read,     write,   add_to_pollset,    add_to_pollset_set,
         shutdown, destroy, get_resource_user, get_peer,
         get_fd};
     grpc_endpoint::vtable = &my_vtable;
     ru_ = grpc_resource_user_create(Library::get().rq(), "dummy_endpoint");
   }

   void PushInput(grpc_exec_ctx *exec_ctx, grpc_slice slice) {
     if (read_cb_ == nullptr) {
       GPR_ASSERT(!have_slice_);
       buffered_slice_ = slice;
       have_slice_ = true;
       return;
     }
     grpc_slice_buffer_add(slices_, slice);
     GRPC_CLOSURE_SCHED(exec_ctx, read_cb_, GRPC_ERROR_NONE);
     read_cb_ = nullptr;
   }

  private:
   grpc_resource_user *ru_;
   grpc_closure *read_cb_ = nullptr;
   grpc_slice_buffer *slices_ = nullptr;
   bool have_slice_ = false;
   grpc_slice buffered_slice_;

   void QueueRead(grpc_exec_ctx *exec_ctx, grpc_slice_buffer *slices,
                  grpc_closure *cb) {
     GPR_ASSERT(read_cb_ == nullptr);
     if (have_slice_) {
       have_slice_ = false;
       grpc_slice_buffer_add(slices, buffered_slice_);
       GRPC_CLOSURE_SCHED(exec_ctx, cb, GRPC_ERROR_NONE);
       return;
     }
     read_cb_ = cb;
     slices_ = slices;
   }

   static void read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                    grpc_slice_buffer *slices, grpc_closure *cb) {
     static_cast<DummyEndpoint *>(ep)->QueueRead(exec_ctx, slices, cb);
   }

   static void write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                     grpc_slice_buffer *slices, grpc_closure *cb) {
     GRPC_CLOSURE_SCHED(exec_ctx, cb, GRPC_ERROR_NONE);
   }

   static grpc_workqueue *get_workqueue(grpc_endpoint *ep) { return NULL; }

   static void add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                              grpc_pollset *pollset) {}

   static void add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                                  grpc_pollset_set *pollset) {}

   static void shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                        grpc_error *why) {
     grpc_resource_user_shutdown(exec_ctx,
                                 static_cast<DummyEndpoint *>(ep)->ru_);
     GRPC_CLOSURE_SCHED(exec_ctx, static_cast<DummyEndpoint *>(ep)->read_cb_,
                        why);
   }

   static void destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
     grpc_resource_user_unref(exec_ctx, static_cast<DummyEndpoint *>(ep)->ru_);
     delete static_cast<DummyEndpoint *>(ep);
   }

   static grpc_resource_user *get_resource_user(grpc_endpoint *ep) {
     return static_cast<DummyEndpoint *>(ep)->ru_;
   }
   static char *get_peer(grpc_endpoint *ep) { return gpr_strdup("test"); }
   static int get_fd(grpc_endpoint *ep) { return 0; }
 };

 class Fixture {
  public:
   Fixture(const grpc::ChannelArguments &args, bool client) {
     grpc_channel_args c_args = args.c_channel_args();
     ep_ = new DummyEndpoint;
     t_ = grpc_create_chttp2_transport(exec_ctx(), &c_args, ep_, client);
     grpc_chttp2_transport_start_reading(exec_ctx(), t_, NULL, NULL);
     FlushExecCtx();
   }

   void FlushExecCtx() { grpc_exec_ctx_flush(&exec_ctx_); }

   ~Fixture() {
     grpc_transport_destroy(&exec_ctx_, t_);
     grpc_exec_ctx_finish(&exec_ctx_);
   }

   grpc_chttp2_transport *chttp2_transport() {
     return reinterpret_cast<grpc_chttp2_transport *>(t_);
   }
   grpc_transport *transport() { return t_; }
   grpc_exec_ctx *exec_ctx() { return &exec_ctx_; }

   void PushInput(grpc_slice slice) { ep_->PushInput(exec_ctx(), slice); }

  private:
   DummyEndpoint *ep_;
   grpc_exec_ctx exec_ctx_ = GRPC_EXEC_CTX_INIT;
   grpc_transport *t_;
 };

 class Closure : public grpc_closure {
  public:
   virtual ~Closure() {}
 };

 template <class F>
 std::unique_ptr<Closure> MakeClosure(
     F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
   struct C : public Closure {
     C(const F &f, grpc_closure_scheduler *sched) : f_(f) {
       GRPC_CLOSURE_INIT(this, Execute, this, sched);
     }
     F f_;
     static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
       static_cast<C *>(arg)->f_(exec_ctx, error);
     }
   };
   return std::unique_ptr<Closure>(new C(f, sched));
 }

 template <class F>
 grpc_closure *MakeOnceClosure(
     F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
   struct C : public grpc_closure {
     C(const F &f) : f_(f) {}
     F f_;
     static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
       static_cast<C *>(arg)->f_(exec_ctx, error);
       delete static_cast<C *>(arg);
     }
   };
   auto *c = new C{f};
   return GRPC_CLOSURE_INIT(c, C::Execute, c, sched);
 }

 class Stream {
  public:
   Stream(Fixture *f) : f_(f) {
     stream_size_ = grpc_transport_stream_size(f->transport());
     stream_ = gpr_malloc(stream_size_);
     arena_ = gpr_arena_create(4096);
   }

   ~Stream() {
     gpr_event_wait(&done_, gpr_inf_future(GPR_CLOCK_REALTIME));
     gpr_free(stream_);
     gpr_arena_destroy(arena_);
   }

   void Init(benchmark::State &state) {
     GRPC_STREAM_REF_INIT(&refcount_, 1, &Stream::FinishDestroy, this,
                          "test_stream");
     gpr_event_init(&done_);
     memset(stream_, 0, stream_size_);
     if ((state.iterations() & 0xffff) == 0) {
       gpr_arena_destroy(arena_);
       arena_ = gpr_arena_create(4096);
     }
     grpc_transport_init_stream(f_->exec_ctx(), f_->transport(),
                                static_cast<grpc_stream *>(stream_), &refcount_,
                                NULL, arena_);
   }

   void DestroyThen(grpc_exec_ctx *exec_ctx, grpc_closure *closure) {
     destroy_closure_ = closure;
 #ifndef NDEBUG
     grpc_stream_unref(exec_ctx, &refcount_, "DestroyThen");
 #else
     grpc_stream_unref(exec_ctx, &refcount_);
 #endif
   }

   void Op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op_batch *op) {
     grpc_transport_perform_stream_op(exec_ctx, f_->transport(),
                                      static_cast<grpc_stream *>(stream_), op);
   }

   grpc_chttp2_stream *chttp2_stream() {
     return static_cast<grpc_chttp2_stream *>(stream_);
   }

  private:
   static void FinishDestroy(grpc_exec_ctx *exec_ctx, void *arg,
                             grpc_error *error) {
     auto stream = static_cast<Stream *>(arg);
     grpc_transport_destroy_stream(exec_ctx, stream->f_->transport(),
                                   static_cast<grpc_stream *>(stream->stream_),
                                   stream->destroy_closure_);
     gpr_event_set(&stream->done_, (void *)1);
   }

   Fixture *f_;
   grpc_stream_refcount refcount_;
   gpr_arena *arena_;
   size_t stream_size_;
   void *stream_;
   grpc_closure *destroy_closure_ = nullptr;
   gpr_event done_;
 };

 ////////////////////////////////////////////////////////////////////////////////
 // Benchmarks
 //

 static void BM_StreamCreateDestroy(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   grpc_transport_stream_op_batch op;
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op, 0, sizeof(op));
   op.cancel_stream = true;
   op.payload = &op_payload;
   op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
   std::unique_ptr<Closure> next =
       MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
         if (!state.KeepRunning()) return;
         s.Init(state);
         s.Op(exec_ctx, &op);
         s.DestroyThen(exec_ctx, next.get());
       });
   GRPC_CLOSURE_RUN(f.exec_ctx(), next.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
   track_counters.Finish(state);
 }
 BENCHMARK(BM_StreamCreateDestroy);

 class RepresentativeClientInitialMetadata {
  public:
   static std::vector<grpc_mdelem> GetElems(grpc_exec_ctx *exec_ctx) {
     return {
         GRPC_MDELEM_SCHEME_HTTP, GRPC_MDELEM_METHOD_POST,
         grpc_mdelem_from_slices(
             exec_ctx, GRPC_MDSTR_PATH,
             grpc_slice_intern(grpc_slice_from_static_string("/foo/bar"))),
         grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_AUTHORITY,
                                 grpc_slice_intern(grpc_slice_from_static_string(
                                     "foo.test.google.fr:1234"))),
         GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_DEFLATE_COMMA_GZIP,
         GRPC_MDELEM_TE_TRAILERS,
         GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC,
         grpc_mdelem_from_slices(
             exec_ctx, GRPC_MDSTR_USER_AGENT,
             grpc_slice_intern(grpc_slice_from_static_string(
                 "grpc-c/3.0.0-dev (linux; chttp2; green)")))};
   }
 };

 template <class Metadata>
 static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   grpc_transport_stream_op_batch op;
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op_payload, 0, sizeof(op_payload));
   std::unique_ptr<Closure> start;
   std::unique_ptr<Closure> done;

   auto reset_op = [&]() {
     memset(&op, 0, sizeof(op));
     op.payload = &op_payload;
   };

   grpc_metadata_batch b;
   grpc_metadata_batch_init(&b);
   b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   std::vector<grpc_mdelem> elems = Metadata::GetElems(f.exec_ctx());
   std::vector<grpc_linked_mdelem> storage(elems.size());
   for (size_t i = 0; i < elems.size(); i++) {
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "addmd",
         grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }

   f.FlushExecCtx();
   start = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     if (!state.KeepRunning()) return;
     s.Init(state);
     reset_op();
     op.on_complete = done.get();
     op.send_initial_metadata = true;
     op.payload->send_initial_metadata.send_initial_metadata = &b;
     s.Op(exec_ctx, &op);
   });
   done = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     reset_op();
     op.cancel_stream = true;
     op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
     s.Op(exec_ctx, &op);
     s.DestroyThen(exec_ctx, start.get());
   });
   GRPC_CLOSURE_SCHED(f.exec_ctx(), start.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
   grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   track_counters.Finish(state);
 }
 BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
                    RepresentativeClientInitialMetadata);

 static void BM_TransportEmptyOp(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   s.Init(state);
   grpc_transport_stream_op_batch op;
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op_payload, 0, sizeof(op_payload));
   auto reset_op = [&]() {
     memset(&op, 0, sizeof(op));
     op.payload = &op_payload;
   };
   std::unique_ptr<Closure> c =
       MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
         if (!state.KeepRunning()) return;
         reset_op();
         op.on_complete = c.get();
         s.Op(exec_ctx, &op);
       });
   GRPC_CLOSURE_SCHED(f.exec_ctx(), c.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
   reset_op();
   op.cancel_stream = true;
   op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
   s.Op(f.exec_ctx(), &op);
   s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
                                                  grpc_error *error) {}));
   f.FlushExecCtx();
   track_counters.Finish(state);
 }
 BENCHMARK(BM_TransportEmptyOp);

 std::vector<std::unique_ptr<gpr_event>> done_events;

 static void BM_TransportStreamSend(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   auto s = std::unique_ptr<Stream>(new Stream(&f));
   s->Init(state);
   grpc_transport_stream_op_batch op;
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op_payload, 0, sizeof(op_payload));
   auto reset_op = [&]() {
     memset(&op, 0, sizeof(op));
     op.payload = &op_payload;
   };
   grpc_slice_buffer_stream send_stream;
   grpc_slice_buffer send_buffer;
   grpc_slice_buffer_init(&send_buffer);
   grpc_slice_buffer_add(&send_buffer, gpr_slice_malloc(state.range(0)));
   memset(GRPC_SLICE_START_PTR(send_buffer.slices[0]), 0,
          GRPC_SLICE_LENGTH(send_buffer.slices[0]));

   grpc_metadata_batch b;
   grpc_metadata_batch_init(&b);
   b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   std::vector<grpc_mdelem> elems =
       RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
   std::vector<grpc_linked_mdelem> storage(elems.size());
   for (size_t i = 0; i < elems.size(); i++) {
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "addmd",
         grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }

   gpr_event *bm_done = new gpr_event;
   gpr_event_init(bm_done);

   std::unique_ptr<Closure> c =
       MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
         if (!state.KeepRunning()) {
           gpr_event_set(bm_done, (void *)1);
           return;
         }
         // force outgoing window to be yuge
         s->chttp2_stream()->flow_control.remote_window_delta =
             1024 * 1024 * 1024;
         f.chttp2_transport()->flow_control.remote_window = 1024 * 1024 * 1024;
         grpc_slice_buffer_stream_init(&send_stream, &send_buffer, 0);
         reset_op();
         op.on_complete = c.get();
         op.send_message = true;
         op.payload->send_message.send_message = &send_stream.base;
         s->Op(exec_ctx, &op);
       });

   reset_op();
   op.send_initial_metadata = true;
   op.payload->send_initial_metadata.send_initial_metadata = &b;
   op.on_complete = c.get();
   s->Op(f.exec_ctx(), &op);

   f.FlushExecCtx();
   gpr_event_wait(bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
   done_events.emplace_back(bm_done);

   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
   s->Op(f.exec_ctx(), &op);
   s->DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
                                                   grpc_error *error) {}));
   f.FlushExecCtx();
   s.reset();
   track_counters.Finish(state);
   grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   grpc_slice_buffer_destroy(&send_buffer);
 }
 BENCHMARK(BM_TransportStreamSend)->Range(0, 128 * 1024 * 1024);

 #define SLICE_FROM_BUFFER(s) grpc_slice_from_static_buffer(s, sizeof(s) - 1)

 static grpc_slice CreateIncomingDataSlice(size_t length, size_t frame_size) {
   std::queue<char> unframed;

   unframed.push(static_cast<uint8_t>(0));
   unframed.push(static_cast<uint8_t>(length >> 24));
   unframed.push(static_cast<uint8_t>(length >> 16));
   unframed.push(static_cast<uint8_t>(length >> 8));
   unframed.push(static_cast<uint8_t>(length));
   for (size_t i = 0; i < length; i++) {
     unframed.push('a');
   }

   std::vector<char> framed;
   while (unframed.size() > frame_size) {
     // frame size
     framed.push_back(static_cast<uint8_t>(frame_size >> 16));
     framed.push_back(static_cast<uint8_t>(frame_size >> 8));
     framed.push_back(static_cast<uint8_t>(frame_size));
     // data frame
     framed.push_back(0);
     // no flags
     framed.push_back(0);
     // stream id
     framed.push_back(0);
     framed.push_back(0);
     framed.push_back(0);
     framed.push_back(1);
     // frame data
     for (size_t i = 0; i < frame_size; i++) {
       framed.push_back(unframed.front());
       unframed.pop();
     }
   }

   // frame size
   framed.push_back(static_cast<uint8_t>(unframed.size() >> 16));
   framed.push_back(static_cast<uint8_t>(unframed.size() >> 8));
   framed.push_back(static_cast<uint8_t>(unframed.size()));
   // data frame
   framed.push_back(0);
   // no flags
   framed.push_back(0);
   // stream id
   framed.push_back(0);
   framed.push_back(0);
   framed.push_back(0);
   framed.push_back(1);
   while (!unframed.empty()) {
     framed.push_back(unframed.front());
     unframed.pop();
   }

   return grpc_slice_from_copied_buffer(framed.data(), framed.size());
 }

 static void BM_TransportStreamRecv(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   s.Init(state);
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op_payload, 0, sizeof(op_payload));
   grpc_transport_stream_op_batch op;
   grpc_byte_stream *recv_stream;
   grpc_slice incoming_data = CreateIncomingDataSlice(state.range(0), 16384);

   auto reset_op = [&]() {
     memset(&op, 0, sizeof(op));
     op.payload = &op_payload;
   };

   grpc_metadata_batch b;
   grpc_metadata_batch_init(&b);
   grpc_metadata_batch b_recv;
   grpc_metadata_batch_init(&b_recv);
   b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   std::vector<grpc_mdelem> elems =
       RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
   std::vector<grpc_linked_mdelem> storage(elems.size());
   for (size_t i = 0; i < elems.size(); i++) {
     GPR_ASSERT(GRPC_LOG_IF_ERROR(
         "addmd",
         grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }

   std::unique_ptr<Closure> do_nothing =
       MakeClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {});

   uint32_t received;

   std::unique_ptr<Closure> drain_start;
   std::unique_ptr<Closure> drain;
   std::unique_ptr<Closure> drain_continue;
   grpc_slice recv_slice;

   std::unique_ptr<Closure> c = MakeClosure([&](grpc_exec_ctx *exec_ctx,
                                                grpc_error *error) {
     if (!state.KeepRunning()) return;
     // force outgoing window to be yuge
     s.chttp2_stream()->flow_control.local_window_delta = 1024 * 1024 * 1024;
     s.chttp2_stream()->flow_control.announced_window_delta = 1024 * 1024 * 1024;
     f.chttp2_transport()->flow_control.announced_window = 1024 * 1024 * 1024;
     received = 0;
     reset_op();
     op.on_complete = do_nothing.get();
     op.recv_message = true;
     op.payload->recv_message.recv_message = &recv_stream;
     op.payload->recv_message.recv_message_ready = drain_start.get();
     s.Op(exec_ctx, &op);
     f.PushInput(grpc_slice_ref(incoming_data));
   });

   drain_start = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     if (recv_stream == NULL) {
       GPR_ASSERT(!state.KeepRunning());
       return;
     }
     GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
   });

   drain = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     do {
       if (received == recv_stream->length) {
         grpc_byte_stream_destroy(exec_ctx, recv_stream);
         GRPC_CLOSURE_SCHED(exec_ctx, c.get(), GRPC_ERROR_NONE);
         return;
       }
     } while (grpc_byte_stream_next(exec_ctx, recv_stream,
                                    recv_stream->length - received,
                                    drain_continue.get()) &&
              GRPC_ERROR_NONE ==
                  grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice) &&
              (received += GRPC_SLICE_LENGTH(recv_slice),
               grpc_slice_unref_internal(exec_ctx, recv_slice), true));
   });

   drain_continue = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice);
     received += GRPC_SLICE_LENGTH(recv_slice);
     grpc_slice_unref_internal(exec_ctx, recv_slice);
     GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
   });

   reset_op();
   op.send_initial_metadata = true;
   op.payload->send_initial_metadata.send_initial_metadata = &b;
   op.recv_initial_metadata = true;
   op.payload->recv_initial_metadata.recv_initial_metadata = &b_recv;
   op.payload->recv_initial_metadata.recv_initial_metadata_ready =
       do_nothing.get();
   op.on_complete = c.get();
   s.Op(f.exec_ctx(), &op);
   f.PushInput(SLICE_FROM_BUFFER(
       "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
       // Generated using:
       // tools/codegen/core/gen_header_frame.py <
       // test/cpp/microbenchmarks/representative_server_initial_metadata.headers
       "\x00\x00X\x01\x04\x00\x00\x00\x01"
       "\x10\x07:status\x03"
       "200"
       "\x10\x0c"
       "content-type\x10"
       "application/grpc"
       "\x10\x14grpc-accept-encoding\x15identity,deflate,gzip"));

   f.FlushExecCtx();
   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
   s.Op(f.exec_ctx(), &op);
   s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
                                                  grpc_error *error) {}));
   f.FlushExecCtx();
   track_counters.Finish(state);
   grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   grpc_metadata_batch_destroy(f.exec_ctx(), &b_recv);
   grpc_slice_unref(incoming_data);
 }
 BENCHMARK(BM_TransportStreamRecv)->Range(0, 128 * 1024 * 1024);

 BENCHMARK_MAIN();
	/*
	*
	* Copyright 2015 gRPC authors.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*
	*/

	/* Microbenchmarks around CHTTP2 transport operations */

	#include <grpc++/support/channel_arguments.h>
	#include <grpc/support/alloc.h>
	#include <grpc/support/log.h>
	#include <grpc/support/string_util.h>
	#include <string.h>
	#include <memory>
	#include <queue>
	#include <sstream>
	extern "C" {
	#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
	#include "src/core/ext/transport/chttp2/transport/internal.h"
	#include "src/core/lib/iomgr/closure.h"
	#include "src/core/lib/iomgr/resource_quota.h"
	#include "src/core/lib/slice/slice_internal.h"
	#include "src/core/lib/transport/static_metadata.h"
	}
	#include "test/cpp/microbenchmarks/helpers.h"
	#include "third_party/benchmark/include/benchmark/benchmark.h"

	auto &force_library_initialization = Library::get();

	////////////////////////////////////////////////////////////////////////////////
	// Helper classes
	//

	class DummyEndpoint : public grpc_endpoint {
	public:
	DummyEndpoint() {
	static const grpc_endpoint_vtable my_vtable = {
	read, write, add_to_pollset, add_to_pollset_set,
	shutdown, destroy, get_resource_user, get_peer,
	get_fd};
	grpc_endpoint::vtable = &my_vtable;
	ru_ = grpc_resource_user_create(Library::get().rq(), "dummy_endpoint");
	}

	void PushInput(grpc_exec_ctx *exec_ctx, grpc_slice slice) {
	if (read_cb_ == nullptr) {
	GPR_ASSERT(!have_slice_);
	buffered_slice_ = slice;
	have_slice_ = true;
	return;
	}
	grpc_slice_buffer_add(slices_, slice);
	GRPC_CLOSURE_SCHED(exec_ctx, read_cb_, GRPC_ERROR_NONE);
	read_cb_ = nullptr;
	}

	private:
	grpc_resource_user *ru_;
	grpc_closure *read_cb_ = nullptr;
	grpc_slice_buffer *slices_ = nullptr;
	bool have_slice_ = false;
	grpc_slice buffered_slice_;

	void QueueRead(grpc_exec_ctx exec_ctx, grpc_slice_buffer slices,
	grpc_closure *cb) {
	GPR_ASSERT(read_cb_ == nullptr);
	if (have_slice_) {
	have_slice_ = false;
	grpc_slice_buffer_add(slices, buffered_slice_);
	GRPC_CLOSURE_SCHED(exec_ctx, cb, GRPC_ERROR_NONE);
	return;
	}
	read_cb_ = cb;
	slices_ = slices;
	}

	static void read(grpc_exec_ctx exec_ctx, grpc_endpoint ep,
	grpc_slice_buffer slices, grpc_closure cb) {
	static_cast<DummyEndpoint *>(ep)->QueueRead(exec_ctx, slices, cb);
	}

	static void write(grpc_exec_ctx exec_ctx, grpc_endpoint ep,
	grpc_slice_buffer slices, grpc_closure cb) {
	GRPC_CLOSURE_SCHED(exec_ctx, cb, GRPC_ERROR_NONE);
	}

	static grpc_workqueue get_workqueue(grpc_endpoint ep) { return NULL; }

	static void add_to_pollset(grpc_exec_ctx exec_ctx, grpc_endpoint ep,
	grpc_pollset *pollset) {}

	static void add_to_pollset_set(grpc_exec_ctx exec_ctx, grpc_endpoint ep,
	grpc_pollset_set *pollset) {}

	static void shutdown(grpc_exec_ctx exec_ctx, grpc_endpoint ep,
	grpc_error *why) {
	grpc_resource_user_shutdown(exec_ctx,
	static_cast<DummyEndpoint *>(ep)->ru_);
	GRPC_CLOSURE_SCHED(exec_ctx, static_cast<DummyEndpoint *>(ep)->read_cb_,
	why);
	}

	static void destroy(grpc_exec_ctx exec_ctx, grpc_endpoint ep) {
	grpc_resource_user_unref(exec_ctx, static_cast<DummyEndpoint *>(ep)->ru_);
	delete static_cast<DummyEndpoint *>(ep);
	}

	static grpc_resource_user get_resource_user(grpc_endpoint ep) {
	return static_cast<DummyEndpoint *>(ep)->ru_;
	}
	static char get_peer(grpc_endpoint ep) { return gpr_strdup("test"); }
	static int get_fd(grpc_endpoint *ep) { return 0; }
	};

	class Fixture {
	public:
	Fixture(const grpc::ChannelArguments &args, bool client) {
	grpc_channel_args c_args = args.c_channel_args();
	ep_ = new DummyEndpoint;
	t_ = grpc_create_chttp2_transport(exec_ctx(), &c_args, ep_, client);
	grpc_chttp2_transport_start_reading(exec_ctx(), t_, NULL, NULL);
	FlushExecCtx();
	}

	void FlushExecCtx() { grpc_exec_ctx_flush(&exec_ctx_); }

	~Fixture() {
	grpc_transport_destroy(&exec_ctx_, t_);
	grpc_exec_ctx_finish(&exec_ctx_);
	}

	grpc_chttp2_transport *chttp2_transport() {
	return reinterpret_cast<grpc_chttp2_transport *>(t_);
	}
	grpc_transport *transport() { return t_; }
	grpc_exec_ctx *exec_ctx() { return &exec_ctx_; }

	void PushInput(grpc_slice slice) { ep_->PushInput(exec_ctx(), slice); }

	private:
	DummyEndpoint *ep_;
	grpc_exec_ctx exec_ctx_ = GRPC_EXEC_CTX_INIT;
	grpc_transport *t_;
	};

	class Closure : public grpc_closure {
	public:
	virtual ~Closure() {}
	};

	template <class F>
	std::unique_ptr<Closure> MakeClosure(
	F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
	struct C : public Closure {
	C(const F &f, grpc_closure_scheduler *sched) : f_(f) {
	GRPC_CLOSURE_INIT(this, Execute, this, sched);
	}
	F f_;
	static void Execute(grpc_exec_ctx exec_ctx, void arg, grpc_error *error) {
	static_cast<C *>(arg)->f_(exec_ctx, error);
	}
	};
	return std::unique_ptr<Closure>(new C(f, sched));
	}

	template <class F>
	grpc_closure *MakeOnceClosure(
	F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
	struct C : public grpc_closure {
	C(const F &f) : f_(f) {}
	F f_;
	static void Execute(grpc_exec_ctx exec_ctx, void arg, grpc_error *error) {
	static_cast<C *>(arg)->f_(exec_ctx, error);
	delete static_cast<C *>(arg);
	}
	};
	auto *c = new C{f};
	return GRPC_CLOSURE_INIT(c, C::Execute, c, sched);
	}

	class Stream {
	public:
	Stream(Fixture *f) : f_(f) {
	stream_size_ = grpc_transport_stream_size(f->transport());
	stream_ = gpr_malloc(stream_size_);
	arena_ = gpr_arena_create(4096);
	}

	~Stream() {
	gpr_event_wait(&done_, gpr_inf_future(GPR_CLOCK_REALTIME));
	gpr_free(stream_);
	gpr_arena_destroy(arena_);
	}

	void Init(benchmark::State &state) {
	GRPC_STREAM_REF_INIT(&refcount_, 1, &Stream::FinishDestroy, this,
	"test_stream");
	gpr_event_init(&done_);
	memset(stream_, 0, stream_size_);
	if ((state.iterations() & 0xffff) == 0) {
	gpr_arena_destroy(arena_);
	arena_ = gpr_arena_create(4096);
	}
	grpc_transport_init_stream(f_->exec_ctx(), f_->transport(),
	static_cast<grpc_stream *>(stream_), &refcount_,
	NULL, arena_);
	}

	void DestroyThen(grpc_exec_ctx exec_ctx, grpc_closure closure) {
	destroy_closure_ = closure;
	#ifndef NDEBUG
	grpc_stream_unref(exec_ctx, &refcount_, "DestroyThen");
	#else
	grpc_stream_unref(exec_ctx, &refcount_);
	#endif
	}

	void Op(grpc_exec_ctx exec_ctx, grpc_transport_stream_op_batch op) {
	grpc_transport_perform_stream_op(exec_ctx, f_->transport(),
	static_cast<grpc_stream *>(stream_), op);
	}

	grpc_chttp2_stream *chttp2_stream() {
	return static_cast<grpc_chttp2_stream *>(stream_);
	}

	private:
	static void FinishDestroy(grpc_exec_ctx exec_ctx, void arg,
	grpc_error *error) {
	auto stream = static_cast<Stream *>(arg);
	grpc_transport_destroy_stream(exec_ctx, stream->f_->transport(),
	static_cast<grpc_stream *>(stream->stream_),
	stream->destroy_closure_);
	gpr_event_set(&stream->done_, (void *)1);
	}

	Fixture *f_;
	grpc_stream_refcount refcount_;
	gpr_arena *arena_;
	size_t stream_size_;
	void *stream_;
	grpc_closure *destroy_closure_ = nullptr;
	gpr_event done_;
	};

	////////////////////////////////////////////////////////////////////////////////
	// Benchmarks
	//

	static void BM_StreamCreateDestroy(benchmark::State &state) {
	TrackCounters track_counters;
	Fixture f(grpc::ChannelArguments(), true);
	Stream s(&f);
	grpc_transport_stream_op_batch op;
	grpc_transport_stream_op_batch_payload op_payload;
	memset(&op, 0, sizeof(op));
	op.cancel_stream = true;
	op.payload = &op_payload;
	op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
	std::unique_ptr<Closure> next =
	MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	if (!state.KeepRunning()) return;
	s.Init(state);
	s.Op(exec_ctx, &op);
	s.DestroyThen(exec_ctx, next.get());
	});
	GRPC_CLOSURE_RUN(f.exec_ctx(), next.get(), GRPC_ERROR_NONE);
	f.FlushExecCtx();
	track_counters.Finish(state);
	}
	BENCHMARK(BM_StreamCreateDestroy);

	class RepresentativeClientInitialMetadata {
	public:
	static std::vector<grpc_mdelem> GetElems(grpc_exec_ctx *exec_ctx) {
	return {
	GRPC_MDELEM_SCHEME_HTTP, GRPC_MDELEM_METHOD_POST,
	grpc_mdelem_from_slices(
	exec_ctx, GRPC_MDSTR_PATH,
	grpc_slice_intern(grpc_slice_from_static_string("/foo/bar"))),
	grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_AUTHORITY,
	grpc_slice_intern(grpc_slice_from_static_string(
	"foo.test.google.fr:1234"))),
	GRPC_MDELEM_GRPC_ACCEPT_ENCODING_IDENTITY_COMMA_DEFLATE_COMMA_GZIP,
	GRPC_MDELEM_TE_TRAILERS,
	GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC,
	grpc_mdelem_from_slices(
	exec_ctx, GRPC_MDSTR_USER_AGENT,
	grpc_slice_intern(grpc_slice_from_static_string(
	"grpc-c/3.0.0-dev (linux; chttp2; green)")))};
	}
	};

	template <class Metadata>
	static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State &state) {
	TrackCounters track_counters;
	Fixture f(grpc::ChannelArguments(), true);
	Stream s(&f);
	grpc_transport_stream_op_batch op;
	grpc_transport_stream_op_batch_payload op_payload;
	memset(&op_payload, 0, sizeof(op_payload));
	std::unique_ptr<Closure> start;
	std::unique_ptr<Closure> done;

	auto reset_op = [&]() {
	memset(&op, 0, sizeof(op));
	op.payload = &op_payload;
	};

	grpc_metadata_batch b;
	grpc_metadata_batch_init(&b);
	b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
	std::vector<grpc_mdelem> elems = Metadata::GetElems(f.exec_ctx());
	std::vector<grpc_linked_mdelem> storage(elems.size());
	for (size_t i = 0; i < elems.size(); i++) {
	GPR_ASSERT(GRPC_LOG_IF_ERROR(
	"addmd",
	grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
	}

	f.FlushExecCtx();
	start = MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	if (!state.KeepRunning()) return;
	s.Init(state);
	reset_op();
	op.on_complete = done.get();
	op.send_initial_metadata = true;
	op.payload->send_initial_metadata.send_initial_metadata = &b;
	s.Op(exec_ctx, &op);
	});
	done = MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	reset_op();
	op.cancel_stream = true;
	op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
	s.Op(exec_ctx, &op);
	s.DestroyThen(exec_ctx, start.get());
	});
	GRPC_CLOSURE_SCHED(f.exec_ctx(), start.get(), GRPC_ERROR_NONE);
	f.FlushExecCtx();
	grpc_metadata_batch_destroy(f.exec_ctx(), &b);
	track_counters.Finish(state);
	}
	BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
	RepresentativeClientInitialMetadata);

	static void BM_TransportEmptyOp(benchmark::State &state) {
	TrackCounters track_counters;
	Fixture f(grpc::ChannelArguments(), true);
	Stream s(&f);
	s.Init(state);
	grpc_transport_stream_op_batch op;
	grpc_transport_stream_op_batch_payload op_payload;
	memset(&op_payload, 0, sizeof(op_payload));
	auto reset_op = [&]() {
	memset(&op, 0, sizeof(op));
	op.payload = &op_payload;
	};
	std::unique_ptr<Closure> c =
	MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	if (!state.KeepRunning()) return;
	reset_op();
	op.on_complete = c.get();
	s.Op(exec_ctx, &op);
	});
	GRPC_CLOSURE_SCHED(f.exec_ctx(), c.get(), GRPC_ERROR_NONE);
	f.FlushExecCtx();
	reset_op();
	op.cancel_stream = true;
	op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
	s.Op(f.exec_ctx(), &op);
	s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
	grpc_error *error) {}));
	f.FlushExecCtx();
	track_counters.Finish(state);
	}
	BENCHMARK(BM_TransportEmptyOp);

	std::vector<std::unique_ptr<gpr_event>> done_events;

	static void BM_TransportStreamSend(benchmark::State &state) {
	TrackCounters track_counters;
	Fixture f(grpc::ChannelArguments(), true);
	auto s = std::unique_ptr<Stream>(new Stream(&f));
	s->Init(state);
	grpc_transport_stream_op_batch op;
	grpc_transport_stream_op_batch_payload op_payload;
	memset(&op_payload, 0, sizeof(op_payload));
	auto reset_op = [&]() {
	memset(&op, 0, sizeof(op));
	op.payload = &op_payload;
	};
	grpc_slice_buffer_stream send_stream;
	grpc_slice_buffer send_buffer;
	grpc_slice_buffer_init(&send_buffer);
	grpc_slice_buffer_add(&send_buffer, gpr_slice_malloc(state.range(0)));
	memset(GRPC_SLICE_START_PTR(send_buffer.slices[0]), 0,
	GRPC_SLICE_LENGTH(send_buffer.slices[0]));

	grpc_metadata_batch b;
	grpc_metadata_batch_init(&b);
	b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
	std::vector<grpc_mdelem> elems =
	RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
	std::vector<grpc_linked_mdelem> storage(elems.size());
	for (size_t i = 0; i < elems.size(); i++) {
	GPR_ASSERT(GRPC_LOG_IF_ERROR(
	"addmd",
	grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
	}

	gpr_event *bm_done = new gpr_event;
	gpr_event_init(bm_done);

	std::unique_ptr<Closure> c =
	MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	if (!state.KeepRunning()) {
	gpr_event_set(bm_done, (void *)1);
	return;
	}
	// force outgoing window to be yuge
	s->chttp2_stream()->flow_control.remote_window_delta =
	1024 * 1024 * 1024;
	f.chttp2_transport()->flow_control.remote_window = 1024 * 1024 * 1024;
	grpc_slice_buffer_stream_init(&send_stream, &send_buffer, 0);
	reset_op();
	op.on_complete = c.get();
	op.send_message = true;
	op.payload->send_message.send_message = &send_stream.base;
	s->Op(exec_ctx, &op);
	});

	reset_op();
	op.send_initial_metadata = true;
	op.payload->send_initial_metadata.send_initial_metadata = &b;
	op.on_complete = c.get();
	s->Op(f.exec_ctx(), &op);

	f.FlushExecCtx();
	gpr_event_wait(bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
	done_events.emplace_back(bm_done);

	reset_op();
	op.cancel_stream = true;
	op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
	s->Op(f.exec_ctx(), &op);
	s->DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
	grpc_error *error) {}));
	f.FlushExecCtx();
	s.reset();
	track_counters.Finish(state);
	grpc_metadata_batch_destroy(f.exec_ctx(), &b);
	grpc_slice_buffer_destroy(&send_buffer);
	}
	BENCHMARK(BM_TransportStreamSend)->Range(0, 128 * 1024 * 1024);

	#define SLICE_FROM_BUFFER(s) grpc_slice_from_static_buffer(s, sizeof(s) - 1)

	static grpc_slice CreateIncomingDataSlice(size_t length, size_t frame_size) {
	std::queue<char> unframed;

	unframed.push(static_cast<uint8_t>(0));
	unframed.push(static_cast<uint8_t>(length >> 24));
	unframed.push(static_cast<uint8_t>(length >> 16));
	unframed.push(static_cast<uint8_t>(length >> 8));
	unframed.push(static_cast<uint8_t>(length));
	for (size_t i = 0; i < length; i++) {
	unframed.push('a');
	}

	std::vector<char> framed;
	while (unframed.size() > frame_size) {
	// frame size
	framed.push_back(static_cast<uint8_t>(frame_size >> 16));
	framed.push_back(static_cast<uint8_t>(frame_size >> 8));
	framed.push_back(static_cast<uint8_t>(frame_size));
	// data frame
	framed.push_back(0);
	// no flags
	framed.push_back(0);
	// stream id
	framed.push_back(0);
	framed.push_back(0);
	framed.push_back(0);
	framed.push_back(1);
	// frame data
	for (size_t i = 0; i < frame_size; i++) {
	framed.push_back(unframed.front());
	unframed.pop();
	}
	}

	// frame size
	framed.push_back(static_cast<uint8_t>(unframed.size() >> 16));
	framed.push_back(static_cast<uint8_t>(unframed.size() >> 8));
	framed.push_back(static_cast<uint8_t>(unframed.size()));
	// data frame
	framed.push_back(0);
	// no flags
	framed.push_back(0);
	// stream id
	framed.push_back(0);
	framed.push_back(0);
	framed.push_back(0);
	framed.push_back(1);
	while (!unframed.empty()) {
	framed.push_back(unframed.front());
	unframed.pop();
	}

	return grpc_slice_from_copied_buffer(framed.data(), framed.size());
	}

	static void BM_TransportStreamRecv(benchmark::State &state) {
	TrackCounters track_counters;
	Fixture f(grpc::ChannelArguments(), true);
	Stream s(&f);
	s.Init(state);
	grpc_transport_stream_op_batch_payload op_payload;
	memset(&op_payload, 0, sizeof(op_payload));
	grpc_transport_stream_op_batch op;
	grpc_byte_stream *recv_stream;
	grpc_slice incoming_data = CreateIncomingDataSlice(state.range(0), 16384);

	auto reset_op = [&]() {
	memset(&op, 0, sizeof(op));
	op.payload = &op_payload;
	};

	grpc_metadata_batch b;
	grpc_metadata_batch_init(&b);
	grpc_metadata_batch b_recv;
	grpc_metadata_batch_init(&b_recv);
	b.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
	std::vector<grpc_mdelem> elems =
	RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
	std::vector<grpc_linked_mdelem> storage(elems.size());
	for (size_t i = 0; i < elems.size(); i++) {
	GPR_ASSERT(GRPC_LOG_IF_ERROR(
	"addmd",
	grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
	}

	std::unique_ptr<Closure> do_nothing =
	MakeClosure([](grpc_exec_ctx exec_ctx, grpc_error error) {});

	uint32_t received;

	std::unique_ptr<Closure> drain_start;
	std::unique_ptr<Closure> drain;
	std::unique_ptr<Closure> drain_continue;
	grpc_slice recv_slice;

	std::unique_ptr<Closure> c = MakeClosure([&](grpc_exec_ctx *exec_ctx,
	grpc_error *error) {
	if (!state.KeepRunning()) return;
	// force outgoing window to be yuge
	s.chttp2_stream()->flow_control.local_window_delta = 1024 * 1024 * 1024;
	s.chttp2_stream()->flow_control.announced_window_delta = 1024 * 1024 * 1024;
	f.chttp2_transport()->flow_control.announced_window = 1024 * 1024 * 1024;
	received = 0;
	reset_op();
	op.on_complete = do_nothing.get();
	op.recv_message = true;
	op.payload->recv_message.recv_message = &recv_stream;
	op.payload->recv_message.recv_message_ready = drain_start.get();
	s.Op(exec_ctx, &op);
	f.PushInput(grpc_slice_ref(incoming_data));
	});

	drain_start = MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	if (recv_stream == NULL) {
	GPR_ASSERT(!state.KeepRunning());
	return;
	}
	GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
	});

	drain = MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	do {
	if (received == recv_stream->length) {
	grpc_byte_stream_destroy(exec_ctx, recv_stream);
	GRPC_CLOSURE_SCHED(exec_ctx, c.get(), GRPC_ERROR_NONE);
	return;
	}
	} while (grpc_byte_stream_next(exec_ctx, recv_stream,
	recv_stream->length - received,
	drain_continue.get()) &&
	GRPC_ERROR_NONE ==
	grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice) &&
	(received += GRPC_SLICE_LENGTH(recv_slice),
	grpc_slice_unref_internal(exec_ctx, recv_slice), true));
	});

	drain_continue = MakeClosure([&](grpc_exec_ctx exec_ctx, grpc_error error) {
	grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice);
	received += GRPC_SLICE_LENGTH(recv_slice);
	grpc_slice_unref_internal(exec_ctx, recv_slice);
	GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
	});

	reset_op();
	op.send_initial_metadata = true;
	op.payload->send_initial_metadata.send_initial_metadata = &b;
	op.recv_initial_metadata = true;
	op.payload->recv_initial_metadata.recv_initial_metadata = &b_recv;
	op.payload->recv_initial_metadata.recv_initial_metadata_ready =
	do_nothing.get();
	op.on_complete = c.get();
	s.Op(f.exec_ctx(), &op);
	f.PushInput(SLICE_FROM_BUFFER(
	"\x00\x00\x00\x04\x00\x00\x00\x00\x00"
	// Generated using:
	// tools/codegen/core/gen_header_frame.py <
	// test/cpp/microbenchmarks/representative_server_initial_metadata.headers
	"\x00\x00X\x01\x04\x00\x00\x00\x01"
	"\x10\x07:status\x03"
	"200"
	"\x10\x0c"
	"content-type\x10"
	"application/grpc"
	"\x10\x14grpc-accept-encoding\x15identity,deflate,gzip"));

	f.FlushExecCtx();
	reset_op();
	op.cancel_stream = true;
	op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
	s.Op(f.exec_ctx(), &op);
	s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
	grpc_error *error) {}));
	f.FlushExecCtx();
	track_counters.Finish(state);
	grpc_metadata_batch_destroy(f.exec_ctx(), &b);
	grpc_metadata_batch_destroy(f.exec_ctx(), &b_recv);
	grpc_slice_unref(incoming_data);
	}
	BENCHMARK(BM_TransportStreamRecv)->Range(0, 128 * 1024 * 1024);

	BENCHMARK_MAIN();