Revert "All instances of exec_ctx being passed around in src/core removed"

1 files changed, 144 insertions, 126 deletions

diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc
index fcb1677d09..be4da4d0bd 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_slice slice) {
+  void PushInput(grpc_exec_ctx* exec_ctx, 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(read_cb_, GRPC_ERROR_NONE);
+    GRPC_CLOSURE_SCHED(exec_ctx, read_cb_, GRPC_ERROR_NONE);
     read_cb_ = nullptr;
   }
 
@@ -77,45 +77,50 @@ class DummyEndpoint : public grpc_endpoint {
   bool have_slice_ = false;
   grpc_slice buffered_slice_;
 
-  void QueueRead(grpc_slice_buffer* slices, grpc_closure* cb) {
+  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(cb, GRPC_ERROR_NONE);
+      GRPC_CLOSURE_SCHED(exec_ctx, cb, GRPC_ERROR_NONE);
       return;
     }
     read_cb_ = cb;
     slices_ = slices;
   }
 
-  static void read(grpc_endpoint* ep, grpc_slice_buffer* slices,
-                   grpc_closure* cb) {
-    static_cast<DummyEndpoint*>(ep)->QueueRead(slices, cb);
+  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_endpoint* ep, grpc_slice_buffer* slices,
-                    grpc_closure* cb) {
-    GRPC_CLOSURE_SCHED(cb, GRPC_ERROR_NONE);
+  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 nullptr; }
 
-  static void add_to_pollset(grpc_endpoint* ep, grpc_pollset* pollset) {}
+  static void add_to_pollset(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep,
+                             grpc_pollset* pollset) {}
 
-  static void add_to_pollset_set(grpc_endpoint* ep, grpc_pollset_set* pollset) {
-  }
+  static void add_to_pollset_set(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep,
+                                 grpc_pollset_set* pollset) {}
 
-  static void delete_from_pollset_set(grpc_endpoint* ep,
+  static void delete_from_pollset_set(grpc_exec_ctx* exec_ctx,
+                                      grpc_endpoint* ep,
                                       grpc_pollset_set* pollset) {}
 
-  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 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_endpoint* ep) {
-    grpc_resource_user_unref(static_cast<DummyEndpoint*>(ep)->ru_);
+  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);
   }
 
@@ -131,24 +136,29 @@ 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(&c_args, ep_, client);
-    grpc_chttp2_transport_start_reading(t_, nullptr, nullptr);
+    t_ = grpc_create_chttp2_transport(exec_ctx(), &c_args, ep_, client);
+    grpc_chttp2_transport_start_reading(exec_ctx(), t_, nullptr, nullptr);
     FlushExecCtx();
   }
 
-  void FlushExecCtx() { grpc_core::ExecCtx::Get()->Flush(); }
+  void FlushExecCtx() { grpc_exec_ctx_flush(&exec_ctx_); }
 
-  ~Fixture() { grpc_transport_destroy(t_); }
+  ~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(slice); }
+  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_;
 };
 
@@ -165,8 +175,8 @@ std::unique_ptr<Closure> MakeClosure(
       GRPC_CLOSURE_INIT(this, Execute, this, sched);
     }
     F f_;
-    static void Execute(void* arg, grpc_error* error) {
-      static_cast<C*>(arg)->f_(error);
+    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));
@@ -178,8 +188,8 @@ grpc_closure* MakeOnceClosure(
   struct C : public grpc_closure {
     C(const F& f) : f_(f) {}
     F f_;
-    static void Execute(void* arg, grpc_error* error) {
-      static_cast<C*>(arg)->f_(error);
+    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);
     }
   };
@@ -210,22 +220,22 @@ class Stream {
       gpr_arena_destroy(arena_);
       arena_ = gpr_arena_create(4096);
     }
-    grpc_transport_init_stream(f_->transport(),
+    grpc_transport_init_stream(f_->exec_ctx(), f_->transport(),
                                static_cast<grpc_stream*>(stream_), &refcount_,
                                nullptr, arena_);
   }
 
-  void DestroyThen(grpc_closure* closure) {
+  void DestroyThen(grpc_exec_ctx* exec_ctx, grpc_closure* closure) {
     destroy_closure_ = closure;
 #ifndef NDEBUG
-    grpc_stream_unref(&refcount_, "DestroyThen");
+    grpc_stream_unref(exec_ctx, &refcount_, "DestroyThen");
 #else
-    grpc_stream_unref(&refcount_);
+    grpc_stream_unref(exec_ctx, &refcount_);
 #endif
   }
 
-  void Op(grpc_transport_stream_op_batch* op) {
-    grpc_transport_perform_stream_op(f_->transport(),
+  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);
   }
 
@@ -234,9 +244,10 @@ class Stream {
   }
 
  private:
-  static void FinishDestroy(void* arg, grpc_error* error) {
+  static void FinishDestroy(grpc_exec_ctx* exec_ctx, void* arg,
+                            grpc_error* error) {
     auto stream = static_cast<Stream*>(arg);
-    grpc_transport_destroy_stream(stream->f_->transport(),
+    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);
@@ -257,7 +268,6 @@ 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;
@@ -266,13 +276,14 @@ 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_error* error) {
-    if (!state.KeepRunning()) return;
-    s.Init(state);
-    s.Op(&op);
-    s.DestroyThen(next.get());
-  });
-  GRPC_CLOSURE_RUN(next.get(), GRPC_ERROR_NONE);
+  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);
 }
@@ -280,21 +291,21 @@ BENCHMARK(BM_StreamCreateDestroy);
 
 class RepresentativeClientInitialMetadata {
  public:
-  static std::vector<grpc_mdelem> GetElems() {
+  static std::vector<grpc_mdelem> GetElems(grpc_exec_ctx* exec_ctx) {
     return {
         GRPC_MDELEM_SCHEME_HTTP,
         GRPC_MDELEM_METHOD_POST,
         grpc_mdelem_from_slices(
-            GRPC_MDSTR_PATH,
+            exec_ctx, GRPC_MDSTR_PATH,
             grpc_slice_intern(grpc_slice_from_static_string("/foo/bar"))),
-        grpc_mdelem_from_slices(GRPC_MDSTR_AUTHORITY,
+        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(
-            GRPC_MDSTR_USER_AGENT,
+            exec_ctx, GRPC_MDSTR_USER_AGENT,
             grpc_slice_intern(grpc_slice_from_static_string(
                 "grpc-c/3.0.0-dev (linux; chttp2; green)")))};
   }
@@ -303,7 +314,6 @@ 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;
@@ -320,33 +330,34 @@ 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();
+  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(&b, &storage[i], elems[i])));
+        "addmd",
+        grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }
 
   f.FlushExecCtx();
-  start = MakeClosure([&](grpc_error* error) {
+  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(&op);
+    s.Op(exec_ctx, &op);
   });
-  done = MakeClosure([&](grpc_error* error) {
+  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(&op);
-    s.DestroyThen(start.get());
+    s.Op(exec_ctx, &op);
+    s.DestroyThen(exec_ctx, start.get());
   });
-  GRPC_CLOSURE_SCHED(start.get(), GRPC_ERROR_NONE);
+  GRPC_CLOSURE_SCHED(f.exec_ctx(), start.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
-  grpc_metadata_batch_destroy(&b);
+  grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   track_counters.Finish(state);
 }
 BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
@@ -354,7 +365,6 @@ 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);
@@ -365,19 +375,21 @@ static void BM_TransportEmptyOp(benchmark::State& state) {
     memset(&op, 0, sizeof(op));
     op.payload = &op_payload;
   };
-  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);
+  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(&op);
-  s.DestroyThen(MakeOnceClosure([](grpc_error* error) {}));
+  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);
 }
@@ -387,7 +399,6 @@ 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);
@@ -409,37 +420,39 @@ 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();
+      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(&b, &storage[i], elems[i])));
+        "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_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);
-  });
+  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);
+      });
 
   reset_op();
   op.send_initial_metadata = true;
   op.payload->send_initial_metadata.send_initial_metadata = &b;
   op.on_complete = c.get();
-  s->Op(&op);
+  s->Op(f.exec_ctx(), &op);
 
   f.FlushExecCtx();
   gpr_event_wait(bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
@@ -448,12 +461,13 @@ static void BM_TransportStreamSend(benchmark::State& state) {
   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
-  s->Op(&op);
-  s->DestroyThen(MakeOnceClosure([](grpc_error* error) {}));
+  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(&b);
+  grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   grpc_slice_buffer_destroy(&send_buffer);
 }
 BENCHMARK(BM_TransportStreamSend)->Range(0, 128 * 1024 * 1024);
@@ -517,7 +531,6 @@ 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);
@@ -538,14 +551,16 @@ 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();
+      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(&b, &storage[i], elems[i])));
+        "addmd",
+        grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }
 
-  std::unique_ptr<Closure> do_nothing = MakeClosure([](grpc_error* error) {});
+  std::unique_ptr<Closure> do_nothing =
+      MakeClosure([](grpc_exec_ctx* exec_ctx, grpc_error* error) {});
 
   uint32_t received;
 
@@ -554,49 +569,51 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   std::unique_ptr<Closure> drain_continue;
   grpc_slice recv_slice;
 
-  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) {
+  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) {
     if (recv_stream == nullptr) {
       GPR_ASSERT(!state.KeepRunning());
       return;
     }
-    GRPC_CLOSURE_RUN(drain.get(), GRPC_ERROR_NONE);
+    GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
   });
 
-  drain = MakeClosure([&](grpc_error* error) {
+  drain = MakeClosure([&](grpc_exec_ctx* exec_ctx, grpc_error* error) {
     do {
       if (received == recv_stream->length) {
-        grpc_byte_stream_destroy(recv_stream);
-        GRPC_CLOSURE_SCHED(c.get(), GRPC_ERROR_NONE);
+        grpc_byte_stream_destroy(exec_ctx, recv_stream);
+        GRPC_CLOSURE_SCHED(exec_ctx, c.get(), GRPC_ERROR_NONE);
         return;
       }
-    } while (grpc_byte_stream_next(recv_stream, recv_stream->length - received,
+    } while (grpc_byte_stream_next(exec_ctx, recv_stream,
+                                   recv_stream->length - received,
                                    drain_continue.get()) &&
              GRPC_ERROR_NONE ==
-                 grpc_byte_stream_pull(recv_stream, &recv_slice) &&
+                 grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice) &&
              (received += GRPC_SLICE_LENGTH(recv_slice),
-              grpc_slice_unref_internal(recv_slice), true));
+              grpc_slice_unref_internal(exec_ctx, recv_slice), true));
   });
 
-  drain_continue = MakeClosure([&](grpc_error* error) {
-    grpc_byte_stream_pull(recv_stream, &recv_slice);
+  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(recv_slice);
-    GRPC_CLOSURE_RUN(drain.get(), GRPC_ERROR_NONE);
+    grpc_slice_unref_internal(exec_ctx, recv_slice);
+    GRPC_CLOSURE_RUN(exec_ctx, drain.get(), GRPC_ERROR_NONE);
   });
 
   reset_op();
@@ -607,7 +624,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(&op);
+  s.Op(f.exec_ctx(), &op);
   f.PushInput(SLICE_FROM_BUFFER(
       "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
       // Generated using:
@@ -625,12 +642,13 @@ static void BM_TransportStreamRecv(benchmark::State& state) {
   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
-  s.Op(&op);
-  s.DestroyThen(MakeOnceClosure([](grpc_error* error) {}));
+  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(&b);
-  grpc_metadata_batch_destroy(&b_recv);
+  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);