Merge remote-tracking branch 'upstream/master' into fix-stream-compression-config-interface

1 files changed, 126 insertions, 144 deletions

diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc
index be4da4d0bd..fcb1677d09 100644
--- a/test/cpp/microbenchmarks/bm_chttp2_transport.cc
+++ b/test/cpp/microbenchmarks/bm_chttp2_transport.cc
@@ -58,7 +58,7 @@ class DummyEndpoint : public grpc_endpoint {
     ru_ = grpc_resource_user_create(Library::get().rq(), "dummy_endpoint");
   }
 
-  void PushInput(grpc_exec_ctx* exec_ctx, grpc_slice slice) {
+  void PushInput(grpc_slice slice) {
     if (read_cb_ == nullptr) {
       GPR_ASSERT(!have_slice_);
       buffered_slice_ = slice;
@@ -66,7 +66,7 @@ class DummyEndpoint : public grpc_endpoint {
       return;
     }
     grpc_slice_buffer_add(slices_, slice);
-    GRPC_CLOSURE_SCHED(exec_ctx, read_cb_, GRPC_ERROR_NONE);
+    GRPC_CLOSURE_SCHED(read_cb_, GRPC_ERROR_NONE);
     read_cb_ = nullptr;
   }
 
@@ -77,50 +77,45 @@ class DummyEndpoint : public grpc_endpoint {
   bool have_slice_ = false;
   grpc_slice buffered_slice_;
 
-  void QueueRead(grpc_exec_ctx* exec_ctx, grpc_slice_buffer* slices,
-                 grpc_closure* cb) {
+  void QueueRead(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);
+      GRPC_CLOSURE_SCHED(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 read(grpc_endpoint* ep, grpc_slice_buffer* slices,
+                   grpc_closure* cb) {
+    static_cast<DummyEndpoint*>(ep)->QueueRead(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 void write(grpc_endpoint* ep, grpc_slice_buffer* slices,
+                    grpc_closure* cb) {
+    GRPC_CLOSURE_SCHED(cb, GRPC_ERROR_NONE);
   }
 
   static grpc_workqueue* get_workqueue(grpc_endpoint* ep) { return nullptr; }
 
-  static void add_to_pollset(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep,
-                             grpc_pollset* pollset) {}
+  static void add_to_pollset(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 add_to_pollset_set(grpc_endpoint* ep, grpc_pollset_set* pollset) {
+  }
 
-  static void delete_from_pollset_set(grpc_exec_ctx* exec_ctx,
-                                      grpc_endpoint* ep,
+  static void delete_from_pollset_set(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 shutdown(grpc_endpoint* ep, grpc_error* why) {
+    grpc_resource_user_shutdown(static_cast<DummyEndpoint*>(ep)->ru_);
+    GRPC_CLOSURE_SCHED(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_);
+  static void destroy(grpc_endpoint* ep) {
+    grpc_resource_user_unref(static_cast<DummyEndpoint*>(ep)->ru_);
     delete static_cast<DummyEndpoint*>(ep);
   }
 
@@ -136,29 +131,24 @@ class Fixture {
   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_, nullptr, nullptr);
+    t_ = grpc_create_chttp2_transport(&c_args, ep_, client);
+    grpc_chttp2_transport_start_reading(t_, nullptr, nullptr);
     FlushExecCtx();
   }
 
-  void FlushExecCtx() { grpc_exec_ctx_flush(&exec_ctx_); }
+  void FlushExecCtx() { grpc_core::ExecCtx::Get()->Flush(); }
 
-  ~Fixture() {
-    grpc_transport_destroy(&exec_ctx_, t_);
-    grpc_exec_ctx_finish(&exec_ctx_);
-  }
+  ~Fixture() { grpc_transport_destroy(t_); }
 
   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); }
+  void PushInput(grpc_slice slice) { ep_->PushInput(slice); }
 
  private:
   DummyEndpoint* ep_;
-  grpc_exec_ctx exec_ctx_ = GRPC_EXEC_CTX_INIT;
   grpc_transport* t_;
 };
 
@@ -175,8 +165,8 @@ std::unique_ptr<Closure> MakeClosure(
       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);
+    static void Execute(void* arg, grpc_error* error) {
+      static_cast<C*>(arg)->f_(error);
     }
   };
   return std::unique_ptr<Closure>(new C(f, sched));
@@ -188,8 +178,8 @@ grpc_closure* MakeOnceClosure(
   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);
+    static void Execute(void* arg, grpc_error* error) {
+      static_cast<C*>(arg)->f_(error);
       delete static_cast<C*>(arg);
     }
   };
@@ -220,22 +210,22 @@ class Stream {
       gpr_arena_destroy(arena_);
       arena_ = gpr_arena_create(4096);
     }
-    grpc_transport_init_stream(f_->exec_ctx(), f_->transport(),
+    grpc_transport_init_stream(f_->transport(),
                                static_cast<grpc_stream*>(stream_), &refcount_,
                                nullptr, arena_);
   }
 
-  void DestroyThen(grpc_exec_ctx* exec_ctx, grpc_closure* closure) {
+  void DestroyThen(grpc_closure* closure) {
     destroy_closure_ = closure;
 #ifndef NDEBUG
-    grpc_stream_unref(exec_ctx, &refcount_, "DestroyThen");
+    grpc_stream_unref(&refcount_, "DestroyThen");
 #else
-    grpc_stream_unref(exec_ctx, &refcount_);
+    grpc_stream_unref(&refcount_);
 #endif
   }
 
-  void Op(grpc_exec_ctx* exec_ctx, grpc_transport_stream_op_batch* op) {
-    grpc_transport_perform_stream_op(exec_ctx, f_->transport(),
+  void Op(grpc_transport_stream_op_batch* op) {
+    grpc_transport_perform_stream_op(f_->transport(),
                                      static_cast<grpc_stream*>(stream_), op);
   }
 
@@ -244,10 +234,9 @@ class Stream {
   }
 
  private:
-  static void FinishDestroy(grpc_exec_ctx* exec_ctx, void* arg,
-                            grpc_error* error) {
+  static void FinishDestroy(void* arg, grpc_error* error) {
     auto stream = static_cast<Stream*>(arg);
-    grpc_transport_destroy_stream(exec_ctx, stream->f_->transport(),
+    grpc_transport_destroy_stream(stream->f_->transport(),
                                   static_cast<grpc_stream*>(stream->stream_),
                                   stream->destroy_closure_);
     gpr_event_set(&stream->done_, (void*)1);
@@ -268,6 +257,7 @@ class Stream {
 
 static void BM_StreamCreateDestroy(benchmark::State& state) {
   TrackCounters track_counters;
+  grpc_core::ExecCtx exec_ctx;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   grpc_transport_stream_op_batch op;
@@ -276,14 +266,13 @@ static void BM_StreamCreateDestroy(benchmark::State& state) {
   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);
+  std::unique_ptr<Closure> next = MakeClosure([&](grpc_error* error) {
+    if (!state.KeepRunning()) return;
+    s.Init(state);
+    s.Op(&op);
+    s.DestroyThen(next.get());
+  });
+  GRPC_CLOSURE_RUN(next.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
   track_counters.Finish(state);
 }
@@ -291,21 +280,21 @@ BENCHMARK(BM_StreamCreateDestroy);
 
 class RepresentativeClientInitialMetadata {
  public:
-  static std::vector<grpc_mdelem> GetElems(grpc_exec_ctx* exec_ctx) {
+  static std::vector<grpc_mdelem> GetElems() {
     return {
         GRPC_MDELEM_SCHEME_HTTP,
         GRPC_MDELEM_METHOD_POST,
         grpc_mdelem_from_slices(
-            exec_ctx, GRPC_MDSTR_PATH,
+            GRPC_MDSTR_PATH,
             grpc_slice_intern(grpc_slice_from_static_string("/foo/bar"))),
-        grpc_mdelem_from_slices(exec_ctx, GRPC_MDSTR_AUTHORITY,
+        grpc_mdelem_from_slices(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_MDSTR_USER_AGENT,
             grpc_slice_intern(grpc_slice_from_static_string(
                 "grpc-c/3.0.0-dev (linux; chttp2; green)")))};
   }
@@ -314,6 +303,7 @@ class RepresentativeClientInitialMetadata {
 template <class Metadata>
 static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State& state) {
   TrackCounters track_counters;
+  grpc_core::ExecCtx exec_ctx;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   grpc_transport_stream_op_batch op;
@@ -330,34 +320,33 @@ static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State& state) {
   grpc_metadata_batch b;
   grpc_metadata_batch_init(&b);
   b.deadline = GRPC_MILLIS_INF_FUTURE;
-  std::vector<grpc_mdelem> elems = Metadata::GetElems(f.exec_ctx());
+  std::vector<grpc_mdelem> elems = Metadata::GetElems();
   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])));
+        "addmd", grpc_metadata_batch_add_tail(&b, &storage[i], elems[i])));
   }
 
   f.FlushExecCtx();
-  start = MakeClosure([&](grpc_exec_ctx* exec_ctx, grpc_error* error) {
+  start = MakeClosure([&](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);
+    s.Op(&op);
   });
-  done = MakeClosure([&](grpc_exec_ctx* exec_ctx, grpc_error* error) {
+  done = MakeClosure([&](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());
+    s.Op(&op);
+    s.DestroyThen(start.get());
   });
-  GRPC_CLOSURE_SCHED(f.exec_ctx(), start.get(), GRPC_ERROR_NONE);
+  GRPC_CLOSURE_SCHED(start.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
-  grpc_metadata_batch_destroy(f.exec_ctx(), &b);
+  grpc_metadata_batch_destroy(&b);
   track_counters.Finish(state);
 }
 BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
@@ -365,6 +354,7 @@ BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
 
 static void BM_TransportEmptyOp(benchmark::State& state) {
   TrackCounters track_counters;
+  grpc_core::ExecCtx exec_ctx;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   s.Init(state);
@@ -375,21 +365,19 @@ static void BM_TransportEmptyOp(benchmark::State& state) {
     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);
+  std::unique_ptr<Closure> c = MakeClosure([&](grpc_error* error) {
+    if (!state.KeepRunning()) return;
+    reset_op();
+    op.on_complete = c.get();
+    s.Op(&op);
+  });
+  GRPC_CLOSURE_SCHED(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) {}));
+  s.Op(&op);
+  s.DestroyThen(MakeOnceClosure([](grpc_error* error) {}));
   f.FlushExecCtx();
   track_counters.Finish(state);
 }
@@ -399,6 +387,7 @@ std::vector<std::unique_ptr<gpr_event>> done_events;
 
 static void BM_TransportStreamSend(benchmark::State& state) {
   TrackCounters track_counters;
+  grpc_core::ExecCtx exec_ctx;
   Fixture f(grpc::ChannelArguments(), true);
   auto s = std::unique_ptr<Stream>(new Stream(&f));
   s->Init(state);
@@ -420,39 +409,37 @@ static void BM_TransportStreamSend(benchmark::State& state) {
   grpc_metadata_batch_init(&b);
   b.deadline = GRPC_MILLIS_INF_FUTURE;
   std::vector<grpc_mdelem> elems =
-      RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
+      RepresentativeClientInitialMetadata::GetElems();
   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])));
+        "addmd", grpc_metadata_batch_add_tail(&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->TestOnlyForceHugeWindow();
-        f.chttp2_transport()->flow_control->TestOnlyForceHugeWindow();
-        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);
-      });
+  std::unique_ptr<Closure> c = MakeClosure([&](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->TestOnlyForceHugeWindow();
+    f.chttp2_transport()->flow_control->TestOnlyForceHugeWindow();
+    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(&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);
+  s->Op(&op);
 
   f.FlushExecCtx();
   gpr_event_wait(bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
@@ -461,13 +448,12 @@ static void BM_TransportStreamSend(benchmark::State& state) {
   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) {}));
+  s->Op(&op);
+  s->DestroyThen(MakeOnceClosure([](grpc_error* error) {}));
   f.FlushExecCtx();
   s.reset();
   track_counters.Finish(state);
-  grpc_metadata_batch_destroy(f.exec_ctx(), &b);
+  grpc_metadata_batch_destroy(&b);
   grpc_slice_buffer_destroy(&send_buffer);
 }
 BENCHMARK(BM_TransportStreamSend)->Range(0, 128 * 1024 * 1024);
@@ -531,6 +517,7 @@ static grpc_slice CreateIncomingDataSlice(size_t length, size_t frame_size) {
 
 static void BM_TransportStreamRecv(benchmark::State& state) {
   TrackCounters track_counters;
+  grpc_core::ExecCtx exec_ctx;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
   s.Init(state);
@@ -551,16 +538,14 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   grpc_metadata_batch_init(&b_recv);
   b.deadline = GRPC_MILLIS_INF_FUTURE;
   std::vector<grpc_mdelem> elems =
-      RepresentativeClientInitialMetadata::GetElems(f.exec_ctx());
+      RepresentativeClientInitialMetadata::GetElems();
   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])));
+        "addmd", grpc_metadata_batch_add_tail(&b, &storage[i], elems[i])));
   }
 
-  std::unique_ptr<Closure> do_nothing =
-      MakeClosure([](grpc_exec_ctx* exec_ctx, grpc_error* error) {});
+  std::unique_ptr<Closure> do_nothing = MakeClosure([](grpc_error* error) {});
 
   uint32_t received;
 
@@ -569,51 +554,49 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   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->TestOnlyForceHugeWindow();
-        f.chttp2_transport()->flow_control->TestOnlyForceHugeWindow();
-        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) {
+  std::unique_ptr<Closure> c = MakeClosure([&](grpc_error* error) {
+    if (!state.KeepRunning()) return;
+    // force outgoing window to be yuge
+    s.chttp2_stream()->flow_control->TestOnlyForceHugeWindow();
+    f.chttp2_transport()->flow_control->TestOnlyForceHugeWindow();
+    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(&op);
+    f.PushInput(grpc_slice_ref(incoming_data));
+  });
+
+  drain_start = MakeClosure([&](grpc_error* error) {
     if (recv_stream == nullptr) {
       GPR_ASSERT(!state.KeepRunning());
       return;
     }
-    GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
+    GRPC_CLOSURE_RUN(drain.get(), GRPC_ERROR_NONE);
   });
 
-  drain = MakeClosure([&](grpc_exec_ctx* exec_ctx, grpc_error* error) {
+  drain = MakeClosure([&](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);
+        grpc_byte_stream_destroy(recv_stream);
+        GRPC_CLOSURE_SCHED(c.get(), GRPC_ERROR_NONE);
         return;
       }
-    } while (grpc_byte_stream_next(exec_ctx, recv_stream,
-                                   recv_stream->length - received,
+    } while (grpc_byte_stream_next(recv_stream, recv_stream->length - received,
                                    drain_continue.get()) &&
              GRPC_ERROR_NONE ==
-                 grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice) &&
+                 grpc_byte_stream_pull(recv_stream, &recv_slice) &&
              (received += GRPC_SLICE_LENGTH(recv_slice),
-              grpc_slice_unref_internal(exec_ctx, recv_slice), true));
+              grpc_slice_unref_internal(recv_slice), true));
   });
 
-  drain_continue = MakeClosure([&](grpc_exec_ctx* exec_ctx, grpc_error* error) {
-    grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice);
+  drain_continue = MakeClosure([&](grpc_error* error) {
+    grpc_byte_stream_pull(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);
+    grpc_slice_unref_internal(recv_slice);
+    GRPC_CLOSURE_RUN(drain.get(), GRPC_ERROR_NONE);
   });
 
   reset_op();
@@ -624,7 +607,7 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   op.payload->recv_initial_metadata.recv_initial_metadata_ready =
       do_nothing.get();
   op.on_complete = c.get();
-  s.Op(f.exec_ctx(), &op);
+  s.Op(&op);
   f.PushInput(SLICE_FROM_BUFFER(
       "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
       // Generated using:
@@ -642,13 +625,12 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   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) {}));
+  s.Op(&op);
+  s.DestroyThen(MakeOnceClosure([](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_metadata_batch_destroy(&b);
+  grpc_metadata_batch_destroy(&b_recv);
   grpc_slice_unref(incoming_data);
 }
 BENCHMARK(BM_TransportStreamRecv)->Range(0, 128 * 1024 * 1024);