Merge pull request #23116 from grpc/revert-23005-asyncwrite

Revert "Fix StartCall: make corking work and allow concurrent Start*"

include/grpcpp/impl/codegen/client_callback_impl.h

@@ -461,51 +461,76 @@ class ClientCallbackReaderWriterImpl
     // 1. Send initial metadata (unless corked) + recv initial metadata
     // 2. Any read backlog
     // 3. Any write backlog
-    // 4. Recv trailing metadata (unless corked)
+    // 4. Recv trailing metadata, on_completion callback
+    started_ = true;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_, /*can_inline=*/false);
     if (!start_corked_) {
       start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
                                      context_->initial_metadata_flags());
     }
-
+    start_ops_.RecvInitialMetadata(context_);
+    start_ops_.set_core_cq_tag(&start_tag_);
     call_.PerformOps(&start_ops_);
 
-    {
+    // Also set up the read and write tags so that they don't have to be set up
-      grpc::internal::MutexLock lock(&start_mu_);
+    // each time
-
+    write_tag_.Set(call_.call(),
-      if (backlog_.read_ops) {
+                   [this](bool ok) {
-        call_.PerformOps(&read_ops_);
+                     reactor_->OnWriteDone(ok);
-      }
+                     MaybeFinish();
-      if (backlog_.write_ops) {
+                   },
-        call_.PerformOps(&write_ops_);
+                   &write_ops_, /*can_inline=*/false);
-      }
+    write_ops_.set_core_cq_tag(&write_tag_);
-      if (backlog_.writes_done_ops) {
+
-        call_.PerformOps(&writes_done_ops_);
+    read_tag_.Set(call_.call(),
-      }
+                  [this](bool ok) {
-      call_.PerformOps(&finish_ops_);
+                    reactor_->OnReadDone(ok);
-      // The last thing in this critical section is to set started_ so that it
+                    MaybeFinish();
-      // can be used lock-free as well.
+                  },
-      started_.store(true, std::memory_order_release);
+                  &read_ops_, /*can_inline=*/false);
+    read_ops_.set_core_cq_tag(&read_tag_);
+    if (read_ops_at_start_) {
+      call_.PerformOps(&read_ops_);
+    }
+
+    if (write_ops_at_start_) {
+      call_.PerformOps(&write_ops_);
+    }
+
+    if (writes_done_ops_at_start_) {
+      call_.PerformOps(&writes_done_ops_);
     }
-    // MaybeFinish outside the lock to make sure that destruction of this object
+
-    // doesn't take place while holding the lock (which would cause the lock to
+    finish_tag_.Set(call_.call(), [this](bool /*ok*/) { MaybeFinish(); },
-    // be released after destruction)
+                    &finish_ops_, /*can_inline=*/false);
-    this->MaybeFinish();
+    finish_ops_.ClientRecvStatus(context_, &finish_status_);
+    finish_ops_.set_core_cq_tag(&finish_tag_);
+    call_.PerformOps(&finish_ops_);
   }
 
   void Read(Response* msg) override {
     read_ops_.RecvMessage(msg);
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
+    if (started_) {
-      grpc::internal::MutexLock lock(&start_mu_);
+      call_.PerformOps(&read_ops_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
+    } else {
-        backlog_.read_ops = true;
+      read_ops_at_start_ = true;
-        return;
-      }
     }
-    call_.PerformOps(&read_ops_);
   }
 
   void Write(const Request* msg, ::grpc::WriteOptions options) override {
+    if (start_corked_) {
+      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+
     if (options.is_last_message()) {
       options.set_buffer_hint();
       write_ops_.ClientSendClose();
@@ -513,22 +538,18 @@ class ClientCallbackReaderWriterImpl
     // TODO(vjpai): don't assert
     GPR_CODEGEN_ASSERT(write_ops_.SendMessagePtr(msg, options).ok());
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-    if (GPR_UNLIKELY(corked_write_needed_)) {
+    if (started_) {
-      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+      call_.PerformOps(&write_ops_);
-                                     context_->initial_metadata_flags());
+    } else {
-      corked_write_needed_ = false;
+      write_ops_at_start_ = true;
-    }
-
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
-      grpc::internal::MutexLock lock(&start_mu_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
-        backlog_.write_ops = true;
-        return;
-      }
     }
-    call_.PerformOps(&write_ops_);
   }
   void WritesDone() override {
+    if (start_corked_) {
+      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                           context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
     writes_done_ops_.ClientSendClose();
     writes_done_tag_.Set(call_.call(),
                          [this](bool ok) {
@@ -538,19 +559,11 @@ class ClientCallbackReaderWriterImpl
                          &writes_done_ops_, /*can_inline=*/false);
     writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-    if (GPR_UNLIKELY(corked_write_needed_)) {
+    if (started_) {
-      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+      call_.PerformOps(&writes_done_ops_);
-                                           context_->initial_metadata_flags());
+    } else {
-      corked_write_needed_ = false;
+      writes_done_ops_at_start_ = true;
     }
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
-      grpc::internal::MutexLock lock(&start_mu_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
-        backlog_.writes_done_ops = true;
-        return;
-      }
-    }
-    call_.PerformOps(&writes_done_ops_);
   }
 
   void AddHold(int holds) override {
@@ -567,42 +580,8 @@ class ClientCallbackReaderWriterImpl
       : context_(context),
         call_(call),
         reactor_(reactor),
-        start_corked_(context_->initial_metadata_corked_),
+        start_corked_(context_->initial_metadata_corked_) {
-        corked_write_needed_(start_corked_) {
     this->BindReactor(reactor);
-
-    // Set up the unchanging parts of the start, read, and write tags and ops.
-    start_tag_.Set(call_.call(),
-                   [this](bool ok) {
-                     reactor_->OnReadInitialMetadataDone(ok);
-                     MaybeFinish();
-                   },
-                   &start_ops_, /*can_inline=*/false);
-    start_ops_.RecvInitialMetadata(context_);
-    start_ops_.set_core_cq_tag(&start_tag_);
-
-    write_tag_.Set(call_.call(),
-                   [this](bool ok) {
-                     reactor_->OnWriteDone(ok);
-                     MaybeFinish();
-                   },
-                   &write_ops_, /*can_inline=*/false);
-    write_ops_.set_core_cq_tag(&write_tag_);
-
-    read_tag_.Set(call_.call(),
-                  [this](bool ok) {
-                    reactor_->OnReadDone(ok);
-                    MaybeFinish();
-                  },
-                  &read_ops_, /*can_inline=*/false);
-    read_ops_.set_core_cq_tag(&read_tag_);
-
-    // Also set up the Finish tag and op set.
-    finish_tag_.Set(call_.call(), [this](bool /*ok*/) { MaybeFinish(); },
-                    &finish_ops_,
-                    /*can_inline=*/false);
-    finish_ops_.ClientRecvStatus(context_, &finish_status_);
-    finish_ops_.set_core_cq_tag(&finish_tag_);
   }
 
   ::grpc_impl::ClientContext* const context_;
@@ -613,9 +592,7 @@ class ClientCallbackReaderWriterImpl
                             grpc::internal::CallOpRecvInitialMetadata>
       start_ops_;
   grpc::internal::CallbackWithSuccessTag start_tag_;
-  const bool start_corked_;
+  bool start_corked_;
-  bool corked_write_needed_;  // no lock needed since only accessed in
-                              // Write/WritesDone which cannot be concurrent
 
   grpc::internal::CallOpSet<grpc::internal::CallOpClientRecvStatus> finish_ops_;
   grpc::internal::CallbackWithSuccessTag finish_tag_;
@@ -626,27 +603,22 @@ class ClientCallbackReaderWriterImpl
                             grpc::internal::CallOpClientSendClose>
       write_ops_;
   grpc::internal::CallbackWithSuccessTag write_tag_;
+  bool write_ops_at_start_{false};
 
   grpc::internal::CallOpSet<grpc::internal::CallOpSendInitialMetadata,
                             grpc::internal::CallOpClientSendClose>
       writes_done_ops_;
   grpc::internal::CallbackWithSuccessTag writes_done_tag_;
+  bool writes_done_ops_at_start_{false};
 
   grpc::internal::CallOpSet<grpc::internal::CallOpRecvMessage<Response>>
       read_ops_;
   grpc::internal::CallbackWithSuccessTag read_tag_;
+  bool read_ops_at_start_{false};
 
-  struct StartCallBacklog {
+  // Minimum of 2 callbacks to pre-register for start and finish
-    bool write_ops = false;
+  std::atomic<intptr_t> callbacks_outstanding_{2};
-    bool writes_done_ops = false;
+  bool started_{false};
-    bool read_ops = false;
-  };
-  StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
-
-  // Minimum of 3 callbacks to pre-register for start ops, StartCall, and finish
-  std::atomic<intptr_t> callbacks_outstanding_{3};
-  std::atomic_bool started_{false};
-  grpc::internal::Mutex start_mu_;
 };
 
 template <class Request, class Response>
@@ -698,7 +670,8 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
     // This call initiates two batches, plus any backlog, each with a callback
     // 1. Send initial metadata (unless corked) + recv initial metadata
     // 2. Any backlog
-    // 3. Recv trailing metadata
+    // 3. Recv trailing metadata, on_completion callback
+    started_ = true;
 
     start_tag_.Set(call_.call(),
                    [this](bool ok) {
@@ -720,13 +693,8 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
                   },
                   &read_ops_, /*can_inline=*/false);
     read_ops_.set_core_cq_tag(&read_tag_);
-
+    if (read_ops_at_start_) {
-    {
+      call_.PerformOps(&read_ops_);
-      grpc::internal::MutexLock lock(&start_mu_);
-      if (backlog_.read_ops) {
-        call_.PerformOps(&read_ops_);
-      }
-      started_.store(true, std::memory_order_release);
     }
 
     finish_tag_.Set(call_.call(), [this](bool /*ok*/) { MaybeFinish(); },
@@ -739,14 +707,11 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
   void Read(Response* msg) override {
     read_ops_.RecvMessage(msg);
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
+    if (started_) {
-      grpc::internal::MutexLock lock(&start_mu_);
+      call_.PerformOps(&read_ops_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
+    } else {
-        backlog_.read_ops = true;
+      read_ops_at_start_ = true;
-        return;
-      }
     }
-    call_.PerformOps(&read_ops_);
   }
 
   void AddHold(int holds) override {
@@ -787,16 +752,11 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
   grpc::internal::CallOpSet<grpc::internal::CallOpRecvMessage<Response>>
       read_ops_;
   grpc::internal::CallbackWithSuccessTag read_tag_;
-
+  bool read_ops_at_start_{false};
-  struct StartCallBacklog {
-    bool read_ops = false;
-  };
-  StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
 
   // Minimum of 2 callbacks to pre-register for start and finish
   std::atomic<intptr_t> callbacks_outstanding_{2};
-  std::atomic_bool started_{false};
+  bool started_{false};
-  grpc::internal::Mutex start_mu_;
 };
 
 template <class Response>
@@ -849,60 +809,74 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
     // This call initiates two batches, plus any backlog, each with a callback
     // 1. Send initial metadata (unless corked) + recv initial metadata
     // 2. Any backlog
-    // 3. Recv trailing metadata
+    // 3. Recv trailing metadata, on_completion callback
+    started_ = true;
 
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_, /*can_inline=*/false);
     if (!start_corked_) {
       start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
                                      context_->initial_metadata_flags());
     }
+    start_ops_.RecvInitialMetadata(context_);
+    start_ops_.set_core_cq_tag(&start_tag_);
     call_.PerformOps(&start_ops_);
 
-    {
+    // Also set up the read and write tags so that they don't have to be set up
-      grpc::internal::MutexLock lock(&start_mu_);
+    // each time
-
+    write_tag_.Set(call_.call(),
-      if (backlog_.write_ops) {
+                   [this](bool ok) {
-        call_.PerformOps(&write_ops_);
+                     reactor_->OnWriteDone(ok);
-      }
+                     MaybeFinish();
-      if (backlog_.writes_done_ops) {
+                   },
-        call_.PerformOps(&writes_done_ops_);
+                   &write_ops_, /*can_inline=*/false);
-      }
+    write_ops_.set_core_cq_tag(&write_tag_);
-      call_.PerformOps(&finish_ops_);
+
-      // The last thing in this critical section is to set started_ so that it
+    if (write_ops_at_start_) {
-      // can be used lock-free as well.
+      call_.PerformOps(&write_ops_);
-      started_.store(true, std::memory_order_release);
     }
-    // MaybeFinish outside the lock to make sure that destruction of this object
+
-    // doesn't take place while holding the lock (which would cause the lock to
+    if (writes_done_ops_at_start_) {
-    // be released after destruction)
+      call_.PerformOps(&writes_done_ops_);
-    this->MaybeFinish();
+    }
+
+    finish_tag_.Set(call_.call(), [this](bool /*ok*/) { MaybeFinish(); },
+                    &finish_ops_, /*can_inline=*/false);
+    finish_ops_.ClientRecvStatus(context_, &finish_status_);
+    finish_ops_.set_core_cq_tag(&finish_tag_);
+    call_.PerformOps(&finish_ops_);
   }
 
   void Write(const Request* msg, ::grpc::WriteOptions options) override {
-    if (GPR_UNLIKELY(options.is_last_message())) {
+    if (start_corked_) {
+      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+
+    if (options.is_last_message()) {
       options.set_buffer_hint();
       write_ops_.ClientSendClose();
     }
     // TODO(vjpai): don't assert
     GPR_CODEGEN_ASSERT(write_ops_.SendMessagePtr(msg, options).ok());
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-
+    if (started_) {
-    if (GPR_UNLIKELY(corked_write_needed_)) {
+      call_.PerformOps(&write_ops_);
-      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+    } else {
-                                     context_->initial_metadata_flags());
+      write_ops_at_start_ = true;
-      corked_write_needed_ = false;
     }
-
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
-      grpc::internal::MutexLock lock(&start_mu_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
-        backlog_.write_ops = true;
-        return;
-      }
-    }
-    call_.PerformOps(&write_ops_);
   }
-
   void WritesDone() override {
+    if (start_corked_) {
+      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                           context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
     writes_done_ops_.ClientSendClose();
     writes_done_tag_.Set(call_.call(),
                          [this](bool ok) {
@@ -912,21 +886,11 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
                          &writes_done_ops_, /*can_inline=*/false);
     writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
     callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed);
-
+    if (started_) {
-    if (GPR_UNLIKELY(corked_write_needed_)) {
+      call_.PerformOps(&writes_done_ops_);
-      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+    } else {
-                                           context_->initial_metadata_flags());
+      writes_done_ops_at_start_ = true;
-      corked_write_needed_ = false;
-    }
-
-    if (GPR_UNLIKELY(!started_.load(std::memory_order_acquire))) {
-      grpc::internal::MutexLock lock(&start_mu_);
-      if (GPR_LIKELY(!started_.load(std::memory_order_relaxed))) {
-        backlog_.writes_done_ops = true;
-        return;
-      }
     }
-    call_.PerformOps(&writes_done_ops_);
   }
 
   void AddHold(int holds) override {
@@ -945,36 +909,10 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
       : context_(context),
         call_(call),
         reactor_(reactor),
-        start_corked_(context_->initial_metadata_corked_),
+        start_corked_(context_->initial_metadata_corked_) {
-        corked_write_needed_(start_corked_) {
     this->BindReactor(reactor);
-
-    // Set up the unchanging parts of the start and write tags and ops.
-    start_tag_.Set(call_.call(),
-                   [this](bool ok) {
-                     reactor_->OnReadInitialMetadataDone(ok);
-                     MaybeFinish();
-                   },
-                   &start_ops_, /*can_inline=*/false);
-    start_ops_.RecvInitialMetadata(context_);
-    start_ops_.set_core_cq_tag(&start_tag_);
-
-    write_tag_.Set(call_.call(),
-                   [this](bool ok) {
-                     reactor_->OnWriteDone(ok);
-                     MaybeFinish();
-                   },
-                   &write_ops_, /*can_inline=*/false);
-    write_ops_.set_core_cq_tag(&write_tag_);
-
-    // Also set up the Finish tag and op set.
     finish_ops_.RecvMessage(response);
     finish_ops_.AllowNoMessage();
-    finish_tag_.Set(call_.call(), [this](bool /*ok*/) { MaybeFinish(); },
-                    &finish_ops_,
-                    /*can_inline=*/false);
-    finish_ops_.ClientRecvStatus(context_, &finish_status_);
-    finish_ops_.set_core_cq_tag(&finish_tag_);
   }
 
   ::grpc_impl::ClientContext* const context_;
@@ -985,9 +923,7 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
                             grpc::internal::CallOpRecvInitialMetadata>
       start_ops_;
   grpc::internal::CallbackWithSuccessTag start_tag_;
-  const bool start_corked_;
+  bool start_corked_;
-  bool corked_write_needed_;  // no lock needed since only accessed in
-                              // Write/WritesDone which cannot be concurrent
 
   grpc::internal::CallOpSet<grpc::internal::CallOpGenericRecvMessage,
                             grpc::internal::CallOpClientRecvStatus>
@@ -1000,22 +936,17 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
                             grpc::internal::CallOpClientSendClose>
       write_ops_;
   grpc::internal::CallbackWithSuccessTag write_tag_;
+  bool write_ops_at_start_{false};
 
   grpc::internal::CallOpSet<grpc::internal::CallOpSendInitialMetadata,
                             grpc::internal::CallOpClientSendClose>
       writes_done_ops_;
   grpc::internal::CallbackWithSuccessTag writes_done_tag_;
+  bool writes_done_ops_at_start_{false};
 
-  struct StartCallBacklog {
+  // Minimum of 2 callbacks to pre-register for start and finish
-    bool write_ops = false;
+  std::atomic<intptr_t> callbacks_outstanding_{2};
-    bool writes_done_ops = false;
+  bool started_{false};
-  };
-  StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
-
-  // Minimum of 3 callbacks to pre-register for start ops, StartCall, and finish
-  std::atomic<intptr_t> callbacks_outstanding_{3};
-  std::atomic_bool started_{false};
-  grpc::internal::Mutex start_mu_;
 };
 
 template <class Request>
@@ -1054,6 +985,7 @@ class ClientCallbackUnaryImpl final : public ClientCallbackUnary {
     // This call initiates two batches, each with a callback
     // 1. Send initial metadata + write + writes done + recv initial metadata
     // 2. Read message, recv trailing metadata
+    started_ = true;
 
     start_tag_.Set(call_.call(),
                    [this](bool ok) {
@@ -1121,6 +1053,7 @@ class ClientCallbackUnaryImpl final : public ClientCallbackUnary {
 
   // This call will have 2 callbacks: start and finish
   std::atomic<intptr_t> callbacks_outstanding_{2};
+  bool started_{false};
 };
 
 class ClientCallbackUnaryFactory {

test/cpp/end2end/client_callback_end2end_test.cc

@@ -16,6 +16,12 @@
  *
  */
 
+#include <algorithm>
+#include <functional>
+#include <mutex>
+#include <sstream>
+#include <thread>
+
 #include <grpcpp/channel.h>
 #include <grpcpp/client_context.h>
 #include <grpcpp/create_channel.h>
@@ -25,14 +31,6 @@
 #include <grpcpp/server_builder.h>
 #include <grpcpp/server_context.h>
 #include <grpcpp/support/client_callback.h>
-#include <gtest/gtest.h>
-
-#include <algorithm>
-#include <condition_variable>
-#include <functional>
-#include <mutex>
-#include <sstream>
-#include <thread>
 
 #include "src/core/lib/gpr/env.h"
 #include "src/core/lib/iomgr/iomgr.h"
@@ -45,6 +43,8 @@
 #include "test/cpp/util/string_ref_helper.h"
 #include "test/cpp/util/test_credentials_provider.h"
 
+#include <gtest/gtest.h>
+
 // MAYBE_SKIP_TEST is a macro to determine if this particular test configuration
 // should be skipped based on a decision made at SetUp time. In particular, any
 // callback tests can only be run if the iomgr can run in the background or if
@@ -1079,8 +1079,7 @@ class BidiClient
  public:
   BidiClient(grpc::testing::EchoTestService::Stub* stub,
              ServerTryCancelRequestPhase server_try_cancel,
-             int num_msgs_to_send, bool cork_metadata, bool first_write_async,
+             int num_msgs_to_send, ClientCancelInfo client_cancel = {})
-             ClientCancelInfo client_cancel = {})
       : server_try_cancel_(server_try_cancel),
         msgs_to_send_{num_msgs_to_send},
         client_cancel_{client_cancel} {
@@ -1090,9 +1089,8 @@ class BidiClient
                            grpc::to_string(server_try_cancel));
     }
     request_.set_message("Hello fren ");
-    context_.set_initial_metadata_corked(cork_metadata);
     stub->experimental_async()->BidiStream(&context_, this);
-    MaybeAsyncWrite(first_write_async);
+    MaybeWrite();
     StartRead(&response_);
     StartCall();
   }
@@ -1113,10 +1111,6 @@ class BidiClient
     }
   }
   void OnWriteDone(bool ok) override {
-    if (async_write_thread_.joinable()) {
-      async_write_thread_.join();
-      RemoveHold();
-    }
     if (server_try_cancel_ == DO_NOT_CANCEL) {
       EXPECT_TRUE(ok);
     } else if (!ok) {
@@ -1181,26 +1175,6 @@ class BidiClient
   }
 
  private:
-  void MaybeAsyncWrite(bool first_write_async) {
-    if (first_write_async) {
-      // Make sure that we have a write to issue.
-      // TODO(vjpai): Make this work with 0 writes case as well.
-      assert(msgs_to_send_ >= 1);
-
-      AddHold();
-      async_write_thread_ = std::thread([this] {
-        std::unique_lock<std::mutex> lock(async_write_thread_mu_);
-        async_write_thread_cv_.wait(
-            lock, [this] { return async_write_thread_start_; });
-        MaybeWrite();
-      });
-      std::lock_guard<std::mutex> lock(async_write_thread_mu_);
-      async_write_thread_start_ = true;
-      async_write_thread_cv_.notify_one();
-      return;
-    }
-    MaybeWrite();
-  }
   void MaybeWrite() {
     if (client_cancel_.cancel &&
         writes_complete_ == client_cancel_.ops_before_cancel) {
@@ -1222,57 +1196,13 @@ class BidiClient
   std::mutex mu_;
   std::condition_variable cv_;
   bool done_ = false;
-  std::thread async_write_thread_;
-  bool async_write_thread_start_ = false;
-  std::mutex async_write_thread_mu_;
-  std::condition_variable async_write_thread_cv_;
 };
 
 TEST_P(ClientCallbackEnd2endTest, BidiStream) {
   MAYBE_SKIP_TEST;
   ResetStub();
-  BidiClient test(stub_.get(), DO_NOT_CANCEL,
+  BidiClient test{stub_.get(), DO_NOT_CANCEL,
-                  kServerDefaultResponseStreamsToSend,
+                  kServerDefaultResponseStreamsToSend};
-                  /*cork_metadata=*/false, /*first_write_async=*/false);
-  test.Await();
-  // Make sure that the server interceptors were not notified of a cancel
-  if (GetParam().use_interceptors) {
-    EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
-  }
-}
-
-TEST_P(ClientCallbackEnd2endTest, BidiStreamFirstWriteAsync) {
-  MAYBE_SKIP_TEST;
-  ResetStub();
-  BidiClient test(stub_.get(), DO_NOT_CANCEL,
-                  kServerDefaultResponseStreamsToSend,
-                  /*cork_metadata=*/false, /*first_write_async=*/true);
-  test.Await();
-  // Make sure that the server interceptors were not notified of a cancel
-  if (GetParam().use_interceptors) {
-    EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
-  }
-}
-
-TEST_P(ClientCallbackEnd2endTest, BidiStreamCorked) {
-  MAYBE_SKIP_TEST;
-  ResetStub();
-  BidiClient test(stub_.get(), DO_NOT_CANCEL,
-                  kServerDefaultResponseStreamsToSend,
-                  /*cork_metadata=*/true, /*first_write_async=*/false);
-  test.Await();
-  // Make sure that the server interceptors were not notified of a cancel
-  if (GetParam().use_interceptors) {
-    EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
-  }
-}
-
-TEST_P(ClientCallbackEnd2endTest, BidiStreamCorkedFirstWriteAsync) {
-  MAYBE_SKIP_TEST;
-  ResetStub();
-  BidiClient test(stub_.get(), DO_NOT_CANCEL,
-                  kServerDefaultResponseStreamsToSend,
-                  /*cork_metadata=*/true, /*first_write_async=*/true);
   test.Await();
   // Make sure that the server interceptors were not notified of a cancel
   if (GetParam().use_interceptors) {
@@ -1283,10 +1213,8 @@ TEST_P(ClientCallbackEnd2endTest, BidiStreamCorkedFirstWriteAsync) {
 TEST_P(ClientCallbackEnd2endTest, ClientCancelsBidiStream) {
   MAYBE_SKIP_TEST;
   ResetStub();
-  BidiClient test(stub_.get(), DO_NOT_CANCEL,
+  BidiClient test{stub_.get(), DO_NOT_CANCEL,
-                  kServerDefaultResponseStreamsToSend,
+                  kServerDefaultResponseStreamsToSend, ClientCancelInfo{2}};
-                  /*cork_metadata=*/false, /*first_write_async=*/false,
-                  ClientCancelInfo(2));
   test.Await();
   // Make sure that the server interceptors were notified of a cancel
   if (GetParam().use_interceptors) {
@@ -1298,8 +1226,7 @@ TEST_P(ClientCallbackEnd2endTest, ClientCancelsBidiStream) {
 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelBefore) {
   MAYBE_SKIP_TEST;
   ResetStub();
-  BidiClient test(stub_.get(), CANCEL_BEFORE_PROCESSING, /*num_msgs_to_send=*/2,
+  BidiClient test{stub_.get(), CANCEL_BEFORE_PROCESSING, 2};
-                  /*cork_metadata=*/false, /*first_write_async=*/false);
   test.Await();
   // Make sure that the server interceptors were notified
   if (GetParam().use_interceptors) {
@@ -1312,9 +1239,7 @@ TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelBefore) {
 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelDuring) {
   MAYBE_SKIP_TEST;
   ResetStub();
-  BidiClient test(stub_.get(), CANCEL_DURING_PROCESSING,
+  BidiClient test{stub_.get(), CANCEL_DURING_PROCESSING, 10};
-                  /*num_msgs_to_send=*/10, /*cork_metadata=*/false,
-                  /*first_write_async=*/false);
   test.Await();
   // Make sure that the server interceptors were notified
   if (GetParam().use_interceptors) {
@@ -1327,8 +1252,7 @@ TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelDuring) {
 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelAfter) {
   MAYBE_SKIP_TEST;
   ResetStub();
-  BidiClient test(stub_.get(), CANCEL_AFTER_PROCESSING, /*num_msgs_to_send=*/5,
+  BidiClient test{stub_.get(), CANCEL_AFTER_PROCESSING, 5};
-                  /*cork_metadata=*/false, /*first_write_async=*/false);
   test.Await();
   // Make sure that the server interceptors were notified
   if (GetParam().use_interceptors) {