Merge github.com:grpc/grpc into flowctlN

examples/cpp/helloworld/greeter_async_client.cc

@@ -60,11 +60,15 @@ class GreeterClient {
     // Storage for the status of the RPC upon completion.
     Status status;
 
-    // stub_->AsyncSayHello() performs the RPC call, returning an instance we
-    // store in "rpc". Because we are using the asynchronous API, we need to
-    // hold on to the "rpc" instance in order to get updates on the ongoing RPC.
+    // stub_->PrepareAsyncSayHello() creates an RPC object, returning
+    // an instance to store in "call" but does not actually start the RPC
+    // Because we are using the asynchronous API, we need to hold on to
+    // the "call" instance in order to get updates on the ongoing RPC.
     std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc(
-        stub_->AsyncSayHello(&context, request, &cq));
+        stub_->PrepareAsyncSayHello(&context, request, &cq));
+
+    // StartCall initiates the RPC call
+    rpc->StartCall();
 
     // Request that, upon completion of the RPC, "reply" be updated with the
     // server's response; "status" with the indication of whether the operation

examples/cpp/helloworld/greeter_async_client2.cc

@@ -49,11 +49,15 @@ class GreeterClient {
         // Call object to store rpc data
         AsyncClientCall* call = new AsyncClientCall;
 
-        // stub_->AsyncSayHello() performs the RPC call, returning an instance to
-        // store in "call". Because we are using the asynchronous API, we need to
-        // hold on to the "call" instance in order to get updates on the ongoing RPC.
-        call->response_reader = stub_->AsyncSayHello(&call->context, request, &cq_);
-
+        // stub_->PrepareAsyncSayHello() creates an RPC object, returning
+        // an instance to store in "call" but does not actually start the RPC
+        // Because we are using the asynchronous API, we need to hold on to
+        // the "call" instance in order to get updates on the ongoing RPC.
+        call->response_reader =
+            stub_->PrepareAsyncSayHello(&call->context, request, &cq_);
+
+        // StartCall initiates the RPC call
+        call->response_reader->StartCall();
 
         // Request that, upon completion of the RPC, "reply" be updated with the
         // server's response; "status" with the indication of whether the operation

include/grpc++/generic/generic_stub.h

@@ -44,6 +44,13 @@ class GenericStub final {
       ClientContext* context, const grpc::string& method, CompletionQueue* cq,
       void* tag);
 
+  /// Setup a call to a named method \a method using \a context, but don't
+  /// start it. Let it be started explicitly with StartCall and a tag.
+  /// The return value only indicates whether or not registration of the call
+  /// succeeded (i.e. the call won't proceed if the return value is nullptr).
+  std::unique_ptr<GenericClientAsyncReaderWriter> PrepareCall(
+      ClientContext* context, const grpc::string& method, CompletionQueue* cq);
+
  private:
   std::shared_ptr<ChannelInterface> channel_;
 };

include/grpc++/impl/codegen/async_stream.h

@@ -35,6 +35,11 @@ class ClientAsyncStreamingInterface {
  public:
   virtual ~ClientAsyncStreamingInterface() {}
 
+  /// Start the call that was set up by the constructor, but only if the
+  /// constructor was invoked through the "Prepare" API which doesn't actually
+  /// start the call
+  virtual void StartCall(void* tag) = 0;
+
   /// Request notification of the reading of the initial metadata. Completion
   /// will be notified by \a tag on the associated completion queue.
   /// This call is optional, but if it is used, it cannot be used concurrently
@@ -156,20 +161,22 @@ class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface,
 template <class R>
 class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
  public:
-  /// Create a stream and write the first request out.
+  /// Create a stream object.
+  /// Write the first request out if \a start is set.
   /// \a tag will be notified on \a cq when the call has been started and
-  /// \a request has been written out.
+  /// \a request has been written out. If \a start is not set, \a tag must be
+  /// nullptr and the actual call must be initiated by StartCall
   /// Note that \a context will be used to fill in custom initial metadata
   /// used to send to the server when starting the call.
   template <class W>
   static ClientAsyncReader* Create(ChannelInterface* channel,
                                    CompletionQueue* cq, const RpcMethod& method,
                                    ClientContext* context, const W& request,
-                                   void* tag) {
+                                   bool start, void* tag) {
     Call call = channel->CreateCall(method, context, cq);
     return new (g_core_codegen_interface->grpc_call_arena_alloc(
         call.call(), sizeof(ClientAsyncReader)))
-        ClientAsyncReader(call, context, request, tag);
+        ClientAsyncReader(call, context, request, start, tag);
   }
 
   // always allocated against a call arena, no memory free required
@@ -177,6 +184,12 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
     assert(size == sizeof(ClientAsyncReader));
   }
 
+  void StartCall(void* tag) override {
+    assert(!started_);
+    started_ = true;
+    StartCallInternal(tag);
+  }
+
   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata
   /// method for semantics.
   ///
@@ -186,6 +199,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
   ///     calling code can access the received metadata through the
   ///     \a ClientContext.
   void ReadInitialMetadata(void* tag) override {
+    assert(started_);
     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 
     meta_ops_.set_output_tag(tag);
@@ -194,6 +208,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
   }
 
   void Read(R* msg, void* tag) override {
+    assert(started_);
     read_ops_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       read_ops_.RecvInitialMetadata(context_);
@@ -208,6 +223,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
   ///   - the \a ClientContext associated with this call is updated with
   ///     possible initial and trailing metadata received from the server.
   void Finish(Status* status, void* tag) override {
+    assert(started_);
     finish_ops_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       finish_ops_.RecvInitialMetadata(context_);
@@ -219,19 +235,28 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
  private:
   template <class W>
   ClientAsyncReader(Call call, ClientContext* context, const W& request,
-                    void* tag)
-      : context_(context), call_(call) {
-    init_ops_.set_output_tag(tag);
-    init_ops_.SendInitialMetadata(context->send_initial_metadata_,
-                                  context->initial_metadata_flags());
+                    bool start, void* tag)
+      : context_(context), call_(call), started_(start) {
     // TODO(ctiller): don't assert
     GPR_CODEGEN_ASSERT(init_ops_.SendMessage(request).ok());
     init_ops_.ClientSendClose();
+    if (start) {
+      StartCallInternal(tag);
+    } else {
+      assert(tag == nullptr);
+    }
+  }
+
+  void StartCallInternal(void* tag) {
+    init_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+                                  context_->initial_metadata_flags());
+    init_ops_.set_output_tag(tag);
     call_.PerformOps(&init_ops_);
   }
 
   ClientContext* context_;
   Call call_;
+  bool started_;
   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
       init_ops_;
   CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
@@ -257,9 +282,12 @@ class ClientAsyncWriterInterface : public ClientAsyncStreamingInterface,
 template <class W>
 class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
  public:
-  /// Create a stream and write the first request out.
+  /// Create a stream object.
+  /// Start the RPC if \a start is set
   /// \a tag will be notified on \a cq when the call has been started (i.e.
   /// intitial metadata sent) and \a request has been written out.
+  /// If \a start is not set, \a tag must be nullptr and the actual call
+  /// must be initiated by StartCall
   /// Note that \a context will be used to fill in custom initial metadata
   /// used to send to the server when starting the call.
   /// \a response will be filled in with the single expected response
@@ -269,11 +297,11 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   static ClientAsyncWriter* Create(ChannelInterface* channel,
                                    CompletionQueue* cq, const RpcMethod& method,
                                    ClientContext* context, R* response,
-                                   void* tag) {
+                                   bool start, void* tag) {
     Call call = channel->CreateCall(method, context, cq);
     return new (g_core_codegen_interface->grpc_call_arena_alloc(
         call.call(), sizeof(ClientAsyncWriter)))
-        ClientAsyncWriter(call, context, response, tag);
+        ClientAsyncWriter(call, context, response, start, tag);
   }
 
   // always allocated against a call arena, no memory free required
@@ -281,6 +309,12 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
     assert(size == sizeof(ClientAsyncWriter));
   }
 
+  void StartCall(void* tag) override {
+    assert(!started_);
+    started_ = true;
+    StartCallInternal(tag);
+  }
+
   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method for
   /// semantics.
   ///
@@ -289,6 +323,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   ///     associated with this call is updated, and the calling code can access
   ///     the received metadata through the \a ClientContext.
   void ReadInitialMetadata(void* tag) override {
+    assert(started_);
     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 
     meta_ops_.set_output_tag(tag);
@@ -297,6 +332,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   }
 
   void Write(const W& msg, void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     // TODO(ctiller): don't assert
     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
@@ -304,6 +340,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   }
 
   void Write(const W& msg, WriteOptions options, void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     if (options.is_last_message()) {
       options.set_buffer_hint();
@@ -315,6 +352,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   }
 
   void WritesDone(void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     write_ops_.ClientSendClose();
     call_.PerformOps(&write_ops_);
@@ -328,6 +366,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   ///   - attempts to fill in the \a response parameter passed to this class's
   ///     constructor with the server's response message.
   void Finish(Status* status, void* tag) override {
+    assert(started_);
     finish_ops_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       finish_ops_.RecvInitialMetadata(context_);
@@ -338,25 +377,32 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
 
  private:
   template <class R>
-  ClientAsyncWriter(Call call, ClientContext* context, R* response, void* tag)
-      : context_(context), call_(call) {
+  ClientAsyncWriter(Call call, ClientContext* context, R* response, bool start,
+                    void* tag)
+      : context_(context), call_(call), started_(start) {
     finish_ops_.RecvMessage(response);
     finish_ops_.AllowNoMessage();
-    // if corked bit is set in context, we buffer up the initial metadata to
-    // coalesce with later message to be sent. No op is performed.
-    if (context_->initial_metadata_corked_) {
-      write_ops_.SendInitialMetadata(context->send_initial_metadata_,
-                                     context->initial_metadata_flags());
+    if (start) {
+      StartCallInternal(tag);
     } else {
+      assert(tag == nullptr);
+    }
+  }
+
+  void StartCallInternal(void* tag) {
+    write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+                                   context_->initial_metadata_flags());
+    // if corked bit is set in context, we just keep the initial metadata
+    // buffered up to coalesce with later message send. No op is performed.
+    if (!context_->initial_metadata_corked_) {
       write_ops_.set_output_tag(tag);
-      write_ops_.SendInitialMetadata(context->send_initial_metadata_,
-                                     context->initial_metadata_flags());
       call_.PerformOps(&write_ops_);
     }
   }
 
   ClientContext* context_;
   Call call_;
+  bool started_;
   CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
       write_ops_;
@@ -388,20 +434,23 @@ template <class W, class R>
 class ClientAsyncReaderWriter final
     : public ClientAsyncReaderWriterInterface<W, R> {
  public:
-  /// Create a stream and write the first request out.
+  /// Create a stream object.
+  /// Start the RPC request if \a start is set.
   /// \a tag will be notified on \a cq when the call has been started (i.e.
-  /// intitial metadata sent).
+  /// intitial metadata sent). If \a start is not set, \a tag must be
+  /// nullptr and the actual call must be initiated by StartCall
   /// Note that \a context will be used to fill in custom initial metadata
   /// used to send to the server when starting the call.
   static ClientAsyncReaderWriter* Create(ChannelInterface* channel,
                                          CompletionQueue* cq,
                                          const RpcMethod& method,
-                                         ClientContext* context, void* tag) {
+                                         ClientContext* context, bool start,
+                                         void* tag) {
     Call call = channel->CreateCall(method, context, cq);
 
     return new (g_core_codegen_interface->grpc_call_arena_alloc(
         call.call(), sizeof(ClientAsyncReaderWriter)))
-        ClientAsyncReaderWriter(call, context, tag);
+        ClientAsyncReaderWriter(call, context, start, tag);
   }
 
   // always allocated against a call arena, no memory free required
@@ -409,6 +458,12 @@ class ClientAsyncReaderWriter final
     assert(size == sizeof(ClientAsyncReaderWriter));
   }
 
+  void StartCall(void* tag) override {
+    assert(!started_);
+    started_ = true;
+    StartCallInternal(tag);
+  }
+
   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method
   /// for semantics of this method.
   ///
@@ -417,6 +472,7 @@ class ClientAsyncReaderWriter final
   ///     is updated with it, and then the receiving initial metadata can
   ///     be accessed through this \a ClientContext.
   void ReadInitialMetadata(void* tag) override {
+    assert(started_);
     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 
     meta_ops_.set_output_tag(tag);
@@ -425,6 +481,7 @@ class ClientAsyncReaderWriter final
   }
 
   void Read(R* msg, void* tag) override {
+    assert(started_);
     read_ops_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       read_ops_.RecvInitialMetadata(context_);
@@ -434,6 +491,7 @@ class ClientAsyncReaderWriter final
   }
 
   void Write(const W& msg, void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     // TODO(ctiller): don't assert
     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
@@ -441,6 +499,7 @@ class ClientAsyncReaderWriter final
   }
 
   void Write(const W& msg, WriteOptions options, void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     if (options.is_last_message()) {
       options.set_buffer_hint();
@@ -452,6 +511,7 @@ class ClientAsyncReaderWriter final
   }
 
   void WritesDone(void* tag) override {
+    assert(started_);
     write_ops_.set_output_tag(tag);
     write_ops_.ClientSendClose();
     call_.PerformOps(&write_ops_);
@@ -462,6 +522,7 @@ class ClientAsyncReaderWriter final
   ///   - the \a ClientContext associated with this call is updated with
   ///     possible initial and trailing metadata sent from the server.
   void Finish(Status* status, void* tag) override {
+    assert(started_);
     finish_ops_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       finish_ops_.RecvInitialMetadata(context_);
@@ -471,23 +532,30 @@ class ClientAsyncReaderWriter final
   }
 
  private:
-  ClientAsyncReaderWriter(Call call, ClientContext* context, void* tag)
-      : context_(context), call_(call) {
-    if (context_->initial_metadata_corked_) {
-      // if corked bit is set in context, we buffer up the initial metadata to
-      // coalesce with later message to be sent. No op is performed.
-      write_ops_.SendInitialMetadata(context->send_initial_metadata_,
-                                     context->initial_metadata_flags());
+  ClientAsyncReaderWriter(Call call, ClientContext* context, bool start,
+                          void* tag)
+      : context_(context), call_(call), started_(start) {
+    if (start) {
+      StartCallInternal(tag);
     } else {
+      assert(tag == nullptr);
+    }
+  }
+
+  void StartCallInternal(void* tag) {
+    write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+                                   context_->initial_metadata_flags());
+    // if corked bit is set in context, we just keep the initial metadata
+    // buffered up to coalesce with later message send. No op is performed.
+    if (!context_->initial_metadata_corked_) {
       write_ops_.set_output_tag(tag);
-      write_ops_.SendInitialMetadata(context->send_initial_metadata_,
-                                     context->initial_metadata_flags());
       call_.PerformOps(&write_ops_);
     }
   }
 
   ClientContext* context_;
   Call call_;
+  bool started_;
   CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
   CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>

include/grpc++/impl/codegen/async_unary_call.h

@@ -32,13 +32,18 @@ namespace grpc {
 class CompletionQueue;
 extern CoreCodegenInterface* g_core_codegen_interface;
 
-/// An interface relevant for async client side unary RPCS (which send
+/// An interface relevant for async client side unary RPCs (which send
 /// one request message to a server and receive one response message).
 template <class R>
 class ClientAsyncResponseReaderInterface {
  public:
   virtual ~ClientAsyncResponseReaderInterface() {}
 
+  /// Start the call that was set up by the constructor, but only if the
+  /// constructor was invoked through the "Prepare" API which doesn't actually
+  /// start the call
+  virtual void StartCall() = 0;
+
   /// Request notification of the reading of initial metadata. Completion
   /// will be notified by \a tag on the associated completion queue.
   /// This call is optional, but if it is used, it cannot be used concurrently
@@ -70,9 +75,10 @@ template <class R>
 class ClientAsyncResponseReader final
     : public ClientAsyncResponseReaderInterface<R> {
  public:
-  /// Start a call and write the request out.
+  /// Start a call and write the request out if \a start is set.
   /// \a tag will be notified on \a cq when the call has been started (i.e.
   /// intitial metadata sent) and \a request has been written out.
+  /// If \a start is not set, the actual call must be initiated by StartCall
   /// Note that \a context will be used to fill in custom initial metadata
   /// used to send to the server when starting the call.
   template <class W>
@@ -80,11 +86,11 @@ class ClientAsyncResponseReader final
                                            CompletionQueue* cq,
                                            const RpcMethod& method,
                                            ClientContext* context,
-                                           const W& request) {
+                                           const W& request, bool start) {
     Call call = channel->CreateCall(method, context, cq);
     return new (g_core_codegen_interface->grpc_call_arena_alloc(
         call.call(), sizeof(ClientAsyncResponseReader)))
-        ClientAsyncResponseReader(call, context, request);
+        ClientAsyncResponseReader(call, context, request, start);
   }
 
   // always allocated against a call arena, no memory free required
@@ -92,13 +98,20 @@ class ClientAsyncResponseReader final
     assert(size == sizeof(ClientAsyncResponseReader));
   }
 
+  void StartCall() override {
+    assert(!started_);
+    started_ = true;
+    StartCallInternal();
+  }
+
   /// See \a ClientAsyncResponseReaderInterface::ReadInitialMetadata for
   /// semantics.
   ///
   /// Side effect:
   ///   - the \a ClientContext associated with this call is updated with
   ///     possible initial and trailing metadata sent from the server.
-  void ReadInitialMetadata(void* tag) {
+  void ReadInitialMetadata(void* tag) override {
+    assert(started_);
     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 
     meta_buf.set_output_tag(tag);
@@ -111,7 +124,8 @@ class ClientAsyncResponseReader final
   /// Side effect:
   ///   - the \a ClientContext associated with this call is updated with
   ///     possible initial and trailing metadata sent from the server.
-  void Finish(R* msg, Status* status, void* tag) {
+  void Finish(R* msg, Status* status, void* tag) override {
+    assert(started_);
     finish_buf.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
       finish_buf.RecvInitialMetadata(context_);
@@ -125,15 +139,22 @@ class ClientAsyncResponseReader final
  private:
   ClientContext* const context_;
   Call call_;
+  bool started_;
 
   template <class W>
-  ClientAsyncResponseReader(Call call, ClientContext* context, const W& request)
-      : context_(context), call_(call) {
-    init_buf.SendInitialMetadata(context->send_initial_metadata_,
-                                 context->initial_metadata_flags());
+  ClientAsyncResponseReader(Call call, ClientContext* context, const W& request,
+                            bool start)
+      : context_(context), call_(call), started_(start) {
+    // Bind the metadata at time of StartCallInternal but set up the rest here
     // TODO(ctiller): don't assert
     GPR_CODEGEN_ASSERT(init_buf.SendMessage(request).ok());
     init_buf.ClientSendClose();
+    if (start) StartCallInternal();
+  }
+
+  void StartCallInternal() {
+    init_buf.SendInitialMetadata(context_->send_initial_metadata_,
+                                 context_->initial_metadata_flags());
     call_.PerformOps(&init_buf);
   }
 

src/compiler/cpp_generator.cc

@@ -165,25 +165,37 @@ void PrintHeaderClientMethodInterfaces(
   (*vars)["Request"] = method->input_type_name();
   (*vars)["Response"] = method->output_type_name();
 
+  struct {
+    grpc::string prefix;
+    grpc::string method_params;  // extra arguments to method
+    grpc::string raw_args;       // extra arguments to raw version of method
+  } async_prefixes[] = {{"Async", ", void* tag", ", tag"},
+                        {"PrepareAsync", "", ""}};
+
   if (is_public) {
     if (method->NoStreaming()) {
       printer->Print(
           *vars,
           "virtual ::grpc::Status $Method$(::grpc::ClientContext* context, "
           "const $Request$& request, $Response$* response) = 0;\n");
-      printer->Print(*vars,
-                     "std::unique_ptr< "
-                     "::grpc::ClientAsyncResponseReaderInterface< $Response$>> "
-                     "Async$Method$(::grpc::ClientContext* context, "
-                     "const $Request$& request, "
-                     "::grpc::CompletionQueue* cq) {\n");
-      printer->Indent();
-      printer->Print(*vars,
-                     "return std::unique_ptr< "
-                     "::grpc::ClientAsyncResponseReaderInterface< $Response$>>("
-                     "Async$Method$Raw(context, request, cq));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< "
+            "::grpc::ClientAsyncResponseReaderInterface< $Response$>> "
+            "$AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+            "const $Request$& request, "
+            "::grpc::CompletionQueue* cq) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< "
+            "::grpc::ClientAsyncResponseReaderInterface< $Response$>>("
+            "$AsyncPrefix$$Method$Raw(context, request, cq));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (ClientOnlyStreaming(method)) {
       printer->Print(
           *vars,
@@ -197,19 +209,26 @@ void PrintHeaderClientMethodInterfaces(
           "($Method$Raw(context, response));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(
-          *vars,
-          "std::unique_ptr< ::grpc::ClientAsyncWriterInterface< $Request$>>"
-          " Async$Method$(::grpc::ClientContext* context, $Response$* "
-          "response, "
-          "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(*vars,
-                     "return std::unique_ptr< "
-                     "::grpc::ClientAsyncWriterInterface< $Request$>>("
-                     "Async$Method$Raw(context, response, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< ::grpc::ClientAsyncWriterInterface< $Request$>>"
+            " $AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+            "$Response$* "
+            "response, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(*vars,
+                       "return std::unique_ptr< "
+                       "::grpc::ClientAsyncWriterInterface< $Request$>>("
+                       "$AsyncPrefix$$Method$Raw(context, response, "
+                       "cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (ServerOnlyStreaming(method)) {
       printer->Print(
           *vars,
@@ -223,19 +242,25 @@ void PrintHeaderClientMethodInterfaces(
           "($Method$Raw(context, request));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(
-          *vars,
-          "std::unique_ptr< ::grpc::ClientAsyncReaderInterface< $Response$>> "
-          "Async$Method$("
-          "::grpc::ClientContext* context, const $Request$& request, "
-          "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(*vars,
-                     "return std::unique_ptr< "
-                     "::grpc::ClientAsyncReaderInterface< $Response$>>("
-                     "Async$Method$Raw(context, request, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< ::grpc::ClientAsyncReaderInterface< $Response$>> "
+            "$AsyncPrefix$$Method$("
+            "::grpc::ClientContext* context, const $Request$& request, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< "
+            "::grpc::ClientAsyncReaderInterface< $Response$>>("
+            "$AsyncPrefix$$Method$Raw(context, request, cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (method->BidiStreaming()) {
       printer->Print(*vars,
                      "std::unique_ptr< ::grpc::ClientReaderWriterInterface< "
@@ -249,61 +274,83 @@ void PrintHeaderClientMethodInterfaces(
           "$Method$Raw(context));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(
-          *vars,
-          "std::unique_ptr< "
-          "::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>> "
-          "Async$Method$(::grpc::ClientContext* context, "
-          "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(
-          *vars,
-          "return std::unique_ptr< "
-          "::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>>("
-          "Async$Method$Raw(context, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< "
+            "::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>> "
+            "$AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< "
+            "::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>>("
+            "$AsyncPrefix$$Method$Raw(context, cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     }
   } else {
     if (method->NoStreaming()) {
-      printer->Print(
-          *vars,
-          "virtual ::grpc::ClientAsyncResponseReaderInterface< $Response$>* "
-          "Async$Method$Raw(::grpc::ClientContext* context, "
-          "const $Request$& request, "
-          "::grpc::CompletionQueue* cq) = 0;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        printer->Print(
+            *vars,
+            "virtual ::grpc::ClientAsyncResponseReaderInterface< $Response$>* "
+            "$AsyncPrefix$$Method$Raw(::grpc::ClientContext* context, "
+            "const $Request$& request, "
+            "::grpc::CompletionQueue* cq) = 0;\n");
+      }
     } else if (ClientOnlyStreaming(method)) {
       printer->Print(
           *vars,
           "virtual ::grpc::ClientWriterInterface< $Request$>*"
           " $Method$Raw("
           "::grpc::ClientContext* context, $Response$* response) = 0;\n");
-      printer->Print(*vars,
-                     "virtual ::grpc::ClientAsyncWriterInterface< $Request$>*"
-                     " Async$Method$Raw(::grpc::ClientContext* context, "
-                     "$Response$* response, "
-                     "::grpc::CompletionQueue* cq, void* tag) = 0;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        printer->Print(
+            *vars,
+            "virtual ::grpc::ClientAsyncWriterInterface< $Request$>*"
+            " $AsyncPrefix$$Method$Raw(::grpc::ClientContext* context, "
+            "$Response$* response, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) = 0;\n");
+      }
     } else if (ServerOnlyStreaming(method)) {
       printer->Print(
           *vars,
           "virtual ::grpc::ClientReaderInterface< $Response$>* $Method$Raw("
           "::grpc::ClientContext* context, const $Request$& request) = 0;\n");
-      printer->Print(
-          *vars,
-          "virtual ::grpc::ClientAsyncReaderInterface< $Response$>* "
-          "Async$Method$Raw("
-          "::grpc::ClientContext* context, const $Request$& request, "
-          "::grpc::CompletionQueue* cq, void* tag) = 0;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        printer->Print(
+            *vars,
+            "virtual ::grpc::ClientAsyncReaderInterface< $Response$>* "
+            "$AsyncPrefix$$Method$Raw("
+            "::grpc::ClientContext* context, const $Request$& request, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) = 0;\n");
+      }
     } else if (method->BidiStreaming()) {
       printer->Print(*vars,
                      "virtual ::grpc::ClientReaderWriterInterface< $Request$, "
                      "$Response$>* "
                      "$Method$Raw(::grpc::ClientContext* context) = 0;\n");
-      printer->Print(*vars,
-                     "virtual ::grpc::ClientAsyncReaderWriterInterface< "
-                     "$Request$, $Response$>* "
-                     "Async$Method$Raw(::grpc::ClientContext* context, "
-                     "::grpc::CompletionQueue* cq, void* tag) = 0;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        printer->Print(
+            *vars,
+            "virtual ::grpc::ClientAsyncReaderWriterInterface< "
+            "$Request$, $Response$>* "
+            "$AsyncPrefix$$Method$Raw(::grpc::ClientContext* context, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) = 0;\n");
+      }
     }
   }
 }
@@ -315,25 +362,35 @@ void PrintHeaderClientMethod(grpc_generator::Printer *printer,
   (*vars)["Method"] = method->name();
   (*vars)["Request"] = method->input_type_name();
   (*vars)["Response"] = method->output_type_name();
+  struct {
+    grpc::string prefix;
+    grpc::string method_params;  // extra arguments to method
+    grpc::string raw_args;       // extra arguments to raw version of method
+  } async_prefixes[] = {{"Async", ", void* tag", ", tag"},
+                        {"PrepareAsync", "", ""}};
+
   if (is_public) {
     if (method->NoStreaming()) {
       printer->Print(
           *vars,
           "::grpc::Status $Method$(::grpc::ClientContext* context, "
           "const $Request$& request, $Response$* response) override;\n");
-      printer->Print(
-          *vars,
-          "std::unique_ptr< ::grpc::ClientAsyncResponseReader< $Response$>> "
-          "Async$Method$(::grpc::ClientContext* context, "
-          "const $Request$& request, "
-          "::grpc::CompletionQueue* cq) {\n");
-      printer->Indent();
-      printer->Print(*vars,
-                     "return std::unique_ptr< "
-                     "::grpc::ClientAsyncResponseReader< $Response$>>("
-                     "Async$Method$Raw(context, request, cq));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< ::grpc::ClientAsyncResponseReader< $Response$>> "
+            "$AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+            "const $Request$& request, "
+            "::grpc::CompletionQueue* cq) {\n");
+        printer->Indent();
+        printer->Print(*vars,
+                       "return std::unique_ptr< "
+                       "::grpc::ClientAsyncResponseReader< $Response$>>("
+                       "$AsyncPrefix$$Method$Raw(context, request, cq));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (ClientOnlyStreaming(method)) {
       printer->Print(
           *vars,
@@ -346,18 +403,24 @@ void PrintHeaderClientMethod(grpc_generator::Printer *printer,
                      "($Method$Raw(context, response));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(*vars,
-                     "std::unique_ptr< ::grpc::ClientAsyncWriter< $Request$>>"
-                     " Async$Method$(::grpc::ClientContext* context, "
-                     "$Response$* response, "
-                     "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(
-          *vars,
-          "return std::unique_ptr< ::grpc::ClientAsyncWriter< $Request$>>("
-          "Async$Method$Raw(context, response, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(*vars,
+                       "std::unique_ptr< ::grpc::ClientAsyncWriter< $Request$>>"
+                       " $AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+                       "$Response$* response, "
+                       "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< ::grpc::ClientAsyncWriter< $Request$>>("
+            "$AsyncPrefix$$Method$Raw(context, response, "
+            "cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (ServerOnlyStreaming(method)) {
       printer->Print(
           *vars,
@@ -371,19 +434,24 @@ void PrintHeaderClientMethod(grpc_generator::Printer *printer,
           "($Method$Raw(context, request));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(
-          *vars,
-          "std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>> "
-          "Async$Method$("
-          "::grpc::ClientContext* context, const $Request$& request, "
-          "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(
-          *vars,
-          "return std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>>("
-          "Async$Method$Raw(context, request, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>> "
+            "$AsyncPrefix$$Method$("
+            "::grpc::ClientContext* context, const $Request$& request, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< ::grpc::ClientAsyncReader< $Response$>>("
+            "$AsyncPrefix$$Method$Raw(context, request, cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     } else if (method->BidiStreaming()) {
       printer->Print(
           *vars,
@@ -396,53 +464,80 @@ void PrintHeaderClientMethod(grpc_generator::Printer *printer,
                      "$Method$Raw(context));\n");
       printer->Outdent();
       printer->Print("}\n");
-      printer->Print(*vars,
-                     "std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< "
-                     "$Request$, $Response$>> "
-                     "Async$Method$(::grpc::ClientContext* context, "
-                     "::grpc::CompletionQueue* cq, void* tag) {\n");
-      printer->Indent();
-      printer->Print(*vars,
-                     "return std::unique_ptr< "
-                     "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>>("
-                     "Async$Method$Raw(context, cq, tag));\n");
-      printer->Outdent();
-      printer->Print("}\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(*vars,
+                       "std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< "
+                       "$Request$, $Response$>> "
+                       "$AsyncPrefix$$Method$(::grpc::ClientContext* context, "
+                       "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+        printer->Indent();
+        printer->Print(
+            *vars,
+            "return std::unique_ptr< "
+            "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>>("
+            "$AsyncPrefix$$Method$Raw(context, cq$AsyncRawArgs$));\n");
+        printer->Outdent();
+        printer->Print("}\n");
+      }
     }
   } else {
     if (method->NoStreaming()) {
-      printer->Print(*vars,
-                     "::grpc::ClientAsyncResponseReader< $Response$>* "
-                     "Async$Method$Raw(::grpc::ClientContext* context, "
-                     "const $Request$& request, "
-                     "::grpc::CompletionQueue* cq) override;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        printer->Print(
+            *vars,
+            "::grpc::ClientAsyncResponseReader< $Response$>* "
+            "$AsyncPrefix$$Method$Raw(::grpc::ClientContext* context, "
+            "const $Request$& request, "
+            "::grpc::CompletionQueue* cq) override;\n");
+      }
     } else if (ClientOnlyStreaming(method)) {
       printer->Print(*vars,
                      "::grpc::ClientWriter< $Request$>* $Method$Raw("
                      "::grpc::ClientContext* context, $Response$* response) "
                      "override;\n");
-      printer->Print(*vars,
-                     "::grpc::ClientAsyncWriter< $Request$>* Async$Method$Raw("
-                     "::grpc::ClientContext* context, $Response$* response, "
-                     "::grpc::CompletionQueue* cq, void* tag) override;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "::grpc::ClientAsyncWriter< $Request$>* $AsyncPrefix$$Method$Raw("
+            "::grpc::ClientContext* context, $Response$* response, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) override;\n");
+      }
     } else if (ServerOnlyStreaming(method)) {
       printer->Print(*vars,
                      "::grpc::ClientReader< $Response$>* $Method$Raw("
                      "::grpc::ClientContext* context, const $Request$& request)"
                      " override;\n");
-      printer->Print(
-          *vars,
-          "::grpc::ClientAsyncReader< $Response$>* Async$Method$Raw("
-          "::grpc::ClientContext* context, const $Request$& request, "
-          "::grpc::CompletionQueue* cq, void* tag) override;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "::grpc::ClientAsyncReader< $Response$>* $AsyncPrefix$$Method$Raw("
+            "::grpc::ClientContext* context, const $Request$& request, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) override;\n");
+      }
     } else if (method->BidiStreaming()) {
       printer->Print(*vars,
                      "::grpc::ClientReaderWriter< $Request$, $Response$>* "
                      "$Method$Raw(::grpc::ClientContext* context) override;\n");
-      printer->Print(*vars,
-                     "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
-                     "Async$Method$Raw(::grpc::ClientContext* context, "
-                     "::grpc::CompletionQueue* cq, void* tag) override;\n");
+      for (auto async_prefix : async_prefixes) {
+        (*vars)["AsyncPrefix"] = async_prefix.prefix;
+        (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+        (*vars)["AsyncRawArgs"] = async_prefix.raw_args;
+        printer->Print(
+            *vars,
+            "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
+            "$AsyncPrefix$$Method$Raw(::grpc::ClientContext* context, "
+            "::grpc::CompletionQueue* cq$AsyncMethodParams$) override;\n");
+      }
     }
   }
 }
@@ -1077,6 +1172,13 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
   (*vars)["Method"] = method->name();
   (*vars)["Request"] = method->input_type_name();
   (*vars)["Response"] = method->output_type_name();
+  struct {
+    grpc::string prefix;
+    grpc::string start;          // bool literal expressed as string
+    grpc::string method_params;  // extra arguments to method
+    grpc::string create_args;    // extra arguments to creator
+  } async_prefixes[] = {{"Async", "true", ", void* tag", ", tag"},
+                        {"PrepareAsync", "false", "", ", nullptr"}};
   if (method->NoStreaming()) {
     printer->Print(*vars,
                    "::grpc::Status $ns$$Service$::Stub::$Method$("
@@ -1087,19 +1189,23 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
                    "rpcmethod_$Method$_, "
                    "context, request, response);\n"
                    "}\n\n");
-    printer->Print(
-        *vars,
-        "::grpc::ClientAsyncResponseReader< $Response$>* "
-        "$ns$$Service$::Stub::Async$Method$Raw(::grpc::ClientContext* context, "
-        "const $Request$& request, "
-        "::grpc::CompletionQueue* cq) {\n");
-    printer->Print(*vars,
-                   "  return "
-                   "::grpc::ClientAsyncResponseReader< $Response$>::Create("
-                   "channel_.get(), cq, "
-                   "rpcmethod_$Method$_, "
-                   "context, request);\n"
-                   "}\n\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncStart"] = async_prefix.start;
+      printer->Print(*vars,
+                     "::grpc::ClientAsyncResponseReader< $Response$>* "
+                     "$ns$$Service$::Stub::$AsyncPrefix$$Method$Raw(::grpc::"
+                     "ClientContext* context, "
+                     "const $Request$& request, "
+                     "::grpc::CompletionQueue* cq) {\n");
+      printer->Print(*vars,
+                     "  return "
+                     "::grpc::ClientAsyncResponseReader< $Response$>::Create("
+                     "channel_.get(), cq, "
+                     "rpcmethod_$Method$_, "
+                     "context, request, $AsyncStart$);\n"
+                     "}\n\n");
+    }
   } else if (ClientOnlyStreaming(method)) {
     printer->Print(*vars,
                    "::grpc::ClientWriter< $Request$>* "
@@ -1111,17 +1217,23 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
                    "rpcmethod_$Method$_, "
                    "context, response);\n"
                    "}\n\n");
-    printer->Print(*vars,
-                   "::grpc::ClientAsyncWriter< $Request$>* "
-                   "$ns$$Service$::Stub::Async$Method$Raw("
-                   "::grpc::ClientContext* context, $Response$* response, "
-                   "::grpc::CompletionQueue* cq, void* tag) {\n");
-    printer->Print(*vars,
-                   "  return ::grpc::ClientAsyncWriter< $Request$>::Create("
-                   "channel_.get(), cq, "
-                   "rpcmethod_$Method$_, "
-                   "context, response, tag);\n"
-                   "}\n\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncStart"] = async_prefix.start;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["AsyncCreateArgs"] = async_prefix.create_args;
+      printer->Print(*vars,
+                     "::grpc::ClientAsyncWriter< $Request$>* "
+                     "$ns$$Service$::Stub::$AsyncPrefix$$Method$Raw("
+                     "::grpc::ClientContext* context, $Response$* response, "
+                     "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+      printer->Print(*vars,
+                     "  return ::grpc::ClientAsyncWriter< $Request$>::Create("
+                     "channel_.get(), cq, "
+                     "rpcmethod_$Method$_, "
+                     "context, response, $AsyncStart$$AsyncCreateArgs$);\n"
+                     "}\n\n");
+    }
   } else if (ServerOnlyStreaming(method)) {
     printer->Print(
         *vars,
@@ -1134,17 +1246,24 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
                    "rpcmethod_$Method$_, "
                    "context, request);\n"
                    "}\n\n");
-    printer->Print(*vars,
-                   "::grpc::ClientAsyncReader< $Response$>* "
-                   "$ns$$Service$::Stub::Async$Method$Raw("
-                   "::grpc::ClientContext* context, const $Request$& request, "
-                   "::grpc::CompletionQueue* cq, void* tag) {\n");
-    printer->Print(*vars,
-                   "  return ::grpc::ClientAsyncReader< $Response$>::Create("
-                   "channel_.get(), cq, "
-                   "rpcmethod_$Method$_, "
-                   "context, request, tag);\n"
-                   "}\n\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncStart"] = async_prefix.start;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["AsyncCreateArgs"] = async_prefix.create_args;
+      printer->Print(
+          *vars,
+          "::grpc::ClientAsyncReader< $Response$>* "
+          "$ns$$Service$::Stub::$AsyncPrefix$$Method$Raw("
+          "::grpc::ClientContext* context, const $Request$& request, "
+          "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+      printer->Print(*vars,
+                     "  return ::grpc::ClientAsyncReader< $Response$>::Create("
+                     "channel_.get(), cq, "
+                     "rpcmethod_$Method$_, "
+                     "context, request, $AsyncStart$$AsyncCreateArgs$);\n"
+                     "}\n\n");
+    }
   } else if (method->BidiStreaming()) {
     printer->Print(
         *vars,
@@ -1157,19 +1276,25 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
                    "rpcmethod_$Method$_, "
                    "context);\n"
                    "}\n\n");
-    printer->Print(
-        *vars,
-        "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
-        "$ns$$Service$::Stub::Async$Method$Raw(::grpc::ClientContext* context, "
-        "::grpc::CompletionQueue* cq, void* tag) {\n");
-    printer->Print(
-        *vars,
-        "  return "
-        "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>::Create("
-        "channel_.get(), cq, "
-        "rpcmethod_$Method$_, "
-        "context, tag);\n"
-        "}\n\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncStart"] = async_prefix.start;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["AsyncCreateArgs"] = async_prefix.create_args;
+      printer->Print(*vars,
+                     "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
+                     "$ns$$Service$::Stub::$AsyncPrefix$$Method$Raw(::grpc::"
+                     "ClientContext* context, "
+                     "::grpc::CompletionQueue* cq$AsyncMethodParams$) {\n");
+      printer->Print(
+          *vars,
+          "  return "
+          "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>::Create("
+          "channel_.get(), cq, "
+          "rpcmethod_$Method$_, "
+          "context, $AsyncStart$$AsyncCreateArgs$);\n"
+          "}\n\n");
+    }
   }
 }
 
@@ -1460,50 +1585,79 @@ void PrintMockClientMethods(grpc_generator::Printer *printer,
   (*vars)["Request"] = method->input_type_name();
   (*vars)["Response"] = method->output_type_name();
 
+  struct {
+    grpc::string prefix;
+    grpc::string method_params;  // extra arguments to method
+    int extra_method_param_count;
+  } async_prefixes[] = {{"Async", ", void* tag", 1}, {"PrepareAsync", "", 0}};
+
   if (method->NoStreaming()) {
     printer->Print(
         *vars,
         "MOCK_METHOD3($Method$, ::grpc::Status(::grpc::ClientContext* context, "
         "const $Request$& request, $Response$* response));\n");
-    printer->Print(*vars,
-                   "MOCK_METHOD3(Async$Method$Raw, "
-                   "::grpc::ClientAsyncResponseReaderInterface< $Response$>*"
-                   "(::grpc::ClientContext* context, const $Request$& request, "
-                   "::grpc::CompletionQueue* cq));\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      printer->Print(
+          *vars,
+          "MOCK_METHOD3($AsyncPrefix$$Method$Raw, "
+          "::grpc::ClientAsyncResponseReaderInterface< $Response$>*"
+          "(::grpc::ClientContext* context, const $Request$& request, "
+          "::grpc::CompletionQueue* cq));\n");
+    }
   } else if (ClientOnlyStreaming(method)) {
     printer->Print(
         *vars,
         "MOCK_METHOD2($Method$Raw, "
         "::grpc::ClientWriterInterface< $Request$>*"
         "(::grpc::ClientContext* context, $Response$* response));\n");
-    printer->Print(*vars,
-                   "MOCK_METHOD4(Async$Method$Raw, "
-                   "::grpc::ClientAsyncWriterInterface< $Request$>*"
-                   "(::grpc::ClientContext* context, $Response$* response, "
-                   "::grpc::CompletionQueue* cq, void* tag));\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["MockArgs"] =
+          std::to_string(3 + async_prefix.extra_method_param_count);
+      printer->Print(*vars,
+                     "MOCK_METHOD$MockArgs$($AsyncPrefix$$Method$Raw, "
+                     "::grpc::ClientAsyncWriterInterface< $Request$>*"
+                     "(::grpc::ClientContext* context, $Response$* response, "
+                     "::grpc::CompletionQueue* cq$AsyncMethodParams$));\n");
+    }
   } else if (ServerOnlyStreaming(method)) {
     printer->Print(
         *vars,
         "MOCK_METHOD2($Method$Raw, "
         "::grpc::ClientReaderInterface< $Response$>*"
         "(::grpc::ClientContext* context, const $Request$& request));\n");
-    printer->Print(*vars,
-                   "MOCK_METHOD4(Async$Method$Raw, "
-                   "::grpc::ClientAsyncReaderInterface< $Response$>*"
-                   "(::grpc::ClientContext* context, const $Request$& request, "
-                   "::grpc::CompletionQueue* cq, void* tag));\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["MockArgs"] =
+          std::to_string(3 + async_prefix.extra_method_param_count);
+      printer->Print(
+          *vars,
+          "MOCK_METHOD$MockArgs$($AsyncPrefix$$Method$Raw, "
+          "::grpc::ClientAsyncReaderInterface< $Response$>*"
+          "(::grpc::ClientContext* context, const $Request$& request, "
+          "::grpc::CompletionQueue* cq$AsyncMethodParams$));\n");
+    }
   } else if (method->BidiStreaming()) {
     printer->Print(
         *vars,
         "MOCK_METHOD1($Method$Raw, "
         "::grpc::ClientReaderWriterInterface< $Request$, $Response$>*"
         "(::grpc::ClientContext* context));\n");
-    printer->Print(
-        *vars,
-        "MOCK_METHOD3(Async$Method$Raw, "
-        "::grpc::ClientAsyncReaderWriterInterface<$Request$, $Response$>*"
-        "(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, "
-        "void* tag));\n");
+    for (auto async_prefix : async_prefixes) {
+      (*vars)["AsyncPrefix"] = async_prefix.prefix;
+      (*vars)["AsyncMethodParams"] = async_prefix.method_params;
+      (*vars)["MockArgs"] =
+          std::to_string(2 + async_prefix.extra_method_param_count);
+      printer->Print(
+          *vars,
+          "MOCK_METHOD$MockArgs$($AsyncPrefix$$Method$Raw, "
+          "::grpc::ClientAsyncReaderWriterInterface<$Request$, $Response$>*"
+          "(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq"
+          "$AsyncMethodParams$));\n");
+    }
   }
 }
 

src/cpp/client/generic_stub.cc

@@ -22,14 +22,29 @@
 
 namespace grpc {
 
+namespace {
+std::unique_ptr<GenericClientAsyncReaderWriter> CallInternal(
+    ChannelInterface* channel, ClientContext* context,
+    const grpc::string& method, CompletionQueue* cq, bool start, void* tag) {
+  return std::unique_ptr<GenericClientAsyncReaderWriter>(
+      GenericClientAsyncReaderWriter::Create(
+          channel, cq, RpcMethod(method.c_str(), RpcMethod::BIDI_STREAMING),
+          context, start, tag));
+}
+
+}  // namespace
+
 // begin a call to a named method
 std::unique_ptr<GenericClientAsyncReaderWriter> GenericStub::Call(
     ClientContext* context, const grpc::string& method, CompletionQueue* cq,
     void* tag) {
-  return std::unique_ptr<GenericClientAsyncReaderWriter>(
-      GenericClientAsyncReaderWriter::Create(
-          channel_.get(), cq,
-          RpcMethod(method.c_str(), RpcMethod::BIDI_STREAMING), context, tag));
+  return CallInternal(channel_.get(), context, method, cq, true, tag);
+}
+
+// setup a call to a named method
+std::unique_ptr<GenericClientAsyncReaderWriter> GenericStub::PrepareCall(
+    ClientContext* context, const grpc::string& method, CompletionQueue* cq) {
+  return CallInternal(channel_.get(), context, method, cq, false, nullptr);
 }
 
 }  // namespace grpc

src/objective-c/GRPCClient/GRPCCall.h

@@ -169,6 +169,13 @@ extern id const kGRPCTrailersKey;
  */
 @property (atomic, copy, readwrite) NSString *serverName;
 
+/**
+ * The timeout for the RPC call in seconds. If set to 0, the call will not timeout. If set to
+ * positive, the gRPC call returns with status GRPCErrorCodeDeadlineExceeded if it is not completed
+ * within \a timeout seconds. A negative value is not allowed.
+ */
+@property NSTimeInterval timeout;
+
 /**
  * The container of the request headers of an RPC conforms to this protocol, which is a subset of
  * NSMutableDictionary's interface. It will become a NSMutableDictionary later on.

src/objective-c/GRPCClient/GRPCCall.m

@@ -423,7 +423,8 @@ static NSString * const kBearerPrefix = @"Bearer ";
 
   _wrappedCall = [[GRPCWrappedCall alloc] initWithHost:_host
                                             serverName:_serverName
-                                                  path:_path];
+                                                  path:_path
+                                               timeout:_timeout];
   NSAssert(_wrappedCall, @"Error allocating RPC objects. Low memory?");
 
   [self sendHeaders:_requestHeaders];

src/objective-c/GRPCClient/private/GRPCChannel.h

@@ -63,5 +63,6 @@ struct grpc_channel_credentials;
 
 - (nullable grpc_call *)unmanagedCallWithPath:(nonnull NSString *)path
                                    serverName:(nonnull NSString *)serverName
+                                      timeout:(NSTimeInterval)timeout
                               completionQueue:(nonnull GRPCCompletionQueue *)queue;
 @end

src/objective-c/GRPCClient/private/GRPCChannel.m

@@ -182,18 +182,28 @@ static grpc_channel_args *BuildChannelArgs(NSDictionary *dictionary) {
 
 - (grpc_call *)unmanagedCallWithPath:(NSString *)path
                           serverName:(NSString *)serverName
+                             timeout:(NSTimeInterval)timeout
                      completionQueue:(GRPCCompletionQueue *)queue {
+  GPR_ASSERT(timeout >= 0);
+  if (timeout < 0) {
+    timeout = 0;
+  }
   grpc_slice host_slice;
   if (serverName) {
     host_slice = grpc_slice_from_copied_string(serverName.UTF8String);
   }
   grpc_slice path_slice = grpc_slice_from_copied_string(path.UTF8String);
+  gpr_timespec deadline_ms = timeout == 0 ?
+                                gpr_inf_future(GPR_CLOCK_REALTIME) :
+                                gpr_time_add(
+                                    gpr_now(GPR_CLOCK_MONOTONIC),
+                                    gpr_time_from_millis((int64_t)(timeout * 1000), GPR_TIMESPAN));
   grpc_call *call = grpc_channel_create_call(_unmanagedChannel,
                                              NULL, GRPC_PROPAGATE_DEFAULTS,
                                              queue.unmanagedQueue,
                                              path_slice,
                                              serverName ? &host_slice : NULL,
-                                             gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+                                             deadline_ms, NULL);
   if (serverName) {
     grpc_slice_unref(host_slice);
   }

src/objective-c/GRPCClient/private/GRPCHost.h

@@ -55,6 +55,7 @@ struct grpc_channel_credentials;
 /** Create a grpc_call object to the provided path on this host. */
 - (nullable struct grpc_call *)unmanagedCallWithPath:(NSString *)path
                                           serverName:(NSString *)serverName
+                                             timeout:(NSTimeInterval)timeout
                                      completionQueue:(GRPCCompletionQueue *)queue;
 
 // TODO: There's a race when a new RPC is coming through just as an existing one is getting

src/objective-c/GRPCClient/private/GRPCHost.m

@@ -121,6 +121,7 @@ static GRPCConnectivityMonitor *connectivityMonitor = nil;
 
 - (nullable grpc_call *)unmanagedCallWithPath:(NSString *)path
                                    serverName:(NSString *)serverName
+                                      timeout:(NSTimeInterval)timeout
                               completionQueue:(GRPCCompletionQueue *)queue {
   GRPCChannel *channel;
   // This is racing -[GRPCHost disconnect].
@@ -130,7 +131,10 @@ static GRPCConnectivityMonitor *connectivityMonitor = nil;
     }
     channel = _channel;
   }
-  return [channel unmanagedCallWithPath:path serverName:serverName completionQueue:queue];
+  return [channel unmanagedCallWithPath:path
+                             serverName:serverName
+                                timeout:timeout
+                        completionQueue:queue];
 }
 
 - (BOOL)setTLSPEMRootCerts:(nullable NSString *)pemRootCerts

src/objective-c/GRPCClient/private/GRPCWrappedCall.h

@@ -76,7 +76,8 @@
 
 - (instancetype)initWithHost:(NSString *)host
                   serverName:(NSString *)serverName
-                        path:(NSString *)path NS_DESIGNATED_INITIALIZER;
+                        path:(NSString *)path
+                     timeout:(NSTimeInterval)timeout NS_DESIGNATED_INITIALIZER;
 
 - (void)startBatchWithOperations:(NSArray *)ops errorHandler:(void(^)())errorHandler;
 

src/objective-c/GRPCClient/private/GRPCWrappedCall.m

@@ -238,12 +238,13 @@
 }
 
 - (instancetype)init {
-  return [self initWithHost:nil serverName:nil path:nil];
+  return [self initWithHost:nil serverName:nil path:nil timeout:0];
 }
 
 - (instancetype)initWithHost:(NSString *)host
                   serverName:(NSString *)serverName
-                        path:(NSString *)path {
+                        path:(NSString *)path
+                     timeout:(NSTimeInterval)timeout {
   if (!path || !host) {
     [NSException raise:NSInvalidArgumentException
                 format:@"path and host cannot be nil."];
@@ -255,7 +256,10 @@
     // queue. Currently we use a singleton queue.
     _queue = [GRPCCompletionQueue completionQueue];
 
-    _call = [[GRPCHost hostWithAddress:host] unmanagedCallWithPath:path serverName:serverName completionQueue:_queue];
+    _call = [[GRPCHost hostWithAddress:host] unmanagedCallWithPath:path
+                                                        serverName:serverName
+                                                           timeout:timeout
+                                                   completionQueue:_queue];
     if (_call == NULL) {
       return nil;
     }

src/objective-c/tests/GRPCClientTests.m

@@ -28,6 +28,7 @@
 #import <RemoteTest/Messages.pbobjc.h>
 #import <RxLibrary/GRXWriteable.h>
 #import <RxLibrary/GRXWriter+Immediate.h>
+#import <RxLibrary/GRXBufferedPipe.h>
 
 #define TEST_TIMEOUT 16
 
@@ -39,6 +40,7 @@ static NSString * const kRemoteSSLHost = @"grpc-test.sandbox.googleapis.com";
 static GRPCProtoMethod *kInexistentMethod;
 static GRPCProtoMethod *kEmptyCallMethod;
 static GRPCProtoMethod *kUnaryCallMethod;
+static GRPCProtoMethod *kFullDuplexCallMethod;
 
 /** Observer class for testing that responseMetadata is KVO-compliant */
 @interface PassthroughObserver : NSObject
@@ -106,6 +108,9 @@ static GRPCProtoMethod *kUnaryCallMethod;
   kUnaryCallMethod = [[GRPCProtoMethod alloc] initWithPackage:kPackage
                                                       service:kService
                                                        method:@"UnaryCall"];
+  kFullDuplexCallMethod = [[GRPCProtoMethod alloc] initWithPackage:kPackage
+                                                           service:kService
+                                                            method:@"FullDuplexCall"];
 }
 
 - (void)testConnectionToRemoteServer {
@@ -422,4 +427,26 @@ static GRPCProtoMethod *kUnaryCallMethod;
   [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
 }
 
+- (void)testTimeout {
+  __weak XCTestExpectation *completion = [self expectationWithDescription:@"RPC completed."];
+
+  GRXBufferedPipe *pipe = [GRXBufferedPipe pipe];
+  GRPCCall *call = [[GRPCCall alloc] initWithHost:kHostAddress
+                                             path:kFullDuplexCallMethod.HTTPPath
+                                   requestsWriter:pipe];
+
+  id<GRXWriteable> responsesWriteable = [[GRXWriteable alloc] initWithValueHandler:^(NSData *value) {
+    XCTAssert(0, @"Failure: response received; Expect: no response received.");
+  } completionHandler:^(NSError *errorOrNil) {
+    XCTAssertNotNil(errorOrNil, @"Failure: no error received; Expect: receive deadline exceeded.");
+    XCTAssertEqual(errorOrNil.code, GRPCErrorCodeDeadlineExceeded);
+    [completion fulfill];
+  }];
+
+  call.timeout = 0.001;
+  [call startWithWriteable:responsesWriteable];
+
+  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
+}
+
 @end

src/objective-c/tests/run_tests.sh

@@ -49,7 +49,8 @@ xcodebuild \
     HOST_PORT_REMOTE=grpc-test.sandbox.googleapis.com \
     test \
     | egrep -v "$XCODEBUILD_FILTER" \
-    | egrep -v '^$' -
+    | egrep -v '^$' \
+    | egrep -v "(GPBDictionary|GPBArray)" -
 
 echo "TIME:  $(date)"
 xcodebuild \
@@ -57,7 +58,8 @@ xcodebuild \
     -scheme CoreCronetEnd2EndTests \
     -destination name="iPhone 6" \
     test \
-    | egrep "$XCODEBUILD_FILTER" \
+    | egrep -v "$XCODEBUILD_FILTER" \
+    | egrep -v '^$' \
     | egrep -v "(GPBDictionary|GPBArray)" -
 
 echo "TIME:  $(date)"
@@ -65,7 +67,10 @@ xcodebuild \
     -workspace Tests.xcworkspace \
     -scheme CronetUnitTests \
     -destination name="iPhone 6" \
-    test | xcpretty
+    test \
+    | egrep -v "$XCODEBUILD_FILTER" \
+    | egrep -v '^$' \
+    | egrep -v "(GPBDictionary|GPBArray)" -
 
 echo "TIME:  $(date)"
 xcodebuild \
@@ -74,5 +79,6 @@ xcodebuild \
     -destination name="iPhone 6" \
     HOST_PORT_REMOTE=grpc-test.sandbox.googleapis.com \
     test \
-    | egrep "$XCODEBUILD_FILTER" \
+    | egrep -v "$XCODEBUILD_FILTER" \
+    | egrep -v '^$' \
     | egrep -v "(GPBDictionary|GPBArray)" -

src/python/grpcio_tests/tests/_sanity/_sanity_test.py

@@ -21,24 +21,25 @@ import six
 import tests
 
 
-class Sanity(unittest.TestCase):
+class SanityTest(unittest.TestCase):
+
+    maxDiff = 32768
 
     def testTestsJsonUpToDate(self):
         """Autodiscovers all test suites and checks that tests.json is up to date"""
         loader = tests.Loader()
         loader.loadTestsFromNames(['tests'])
-        test_suite_names = [
+        test_suite_names = sorted({
             test_case_class.id().rsplit('.', 1)[0]
             for test_case_class in tests._loader.iterate_suite_cases(
                 loader.suite)
-        ]
-        test_suite_names = sorted(set(test_suite_names))
+        })
 
         tests_json_string = pkg_resources.resource_string('tests', 'tests.json')
-        if six.PY3:
-            tests_json_string = tests_json_string.decode()
-        tests_json = json.loads(tests_json_string)
-        self.assertListEqual(test_suite_names, tests_json)
+        tests_json = json.loads(tests_json_string.decode()
+                                if six.PY3 else tests_json_string)
+
+        self.assertSequenceEqual(tests_json, test_suite_names)
 
 
 if __name__ == '__main__':

src/python/grpcio_tests/tests/protoc_plugin/_python_plugin_test.py

@@ -33,7 +33,7 @@ from tests.unit.framework.common import test_constants
 import tests.protoc_plugin.protos.payload.test_payload_pb2 as payload_pb2
 import tests.protoc_plugin.protos.requests.r.test_requests_pb2 as request_pb2
 import tests.protoc_plugin.protos.responses.test_responses_pb2 as response_pb2
-import tests.protoc_plugin.protos.service.test_service_pb2 as service_pb2
+import tests.protoc_plugin.protos.service.test_service_pb2_grpc as service_pb2_grpc
 
 # Identifiers of entities we expect to find in the generated module.
 STUB_IDENTIFIER = 'TestServiceStub'
@@ -138,7 +138,7 @@ def _CreateService():
   """
     servicer_methods = _ServicerMethods()
 
-    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+    class Servicer(getattr(service_pb2_grpc, SERVICER_IDENTIFIER)):
 
         def UnaryCall(self, request, context):
             return servicer_methods.UnaryCall(request, context)
@@ -157,11 +157,12 @@ def _CreateService():
 
     server = grpc.server(
         futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-    getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
+    getattr(service_pb2_grpc, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(),
+                                                                 server)
     port = server.add_insecure_port('[::]:0')
     server.start()
     channel = grpc.insecure_channel('localhost:{}'.format(port))
-    stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
+    stub = getattr(service_pb2_grpc, STUB_IDENTIFIER)(channel)
     return _Service(servicer_methods, server, stub)
 
 
@@ -173,16 +174,17 @@ def _CreateIncompleteService():
       servicer_methods implements none of the methods required of it.
   """
 
-    class Servicer(getattr(service_pb2, SERVICER_IDENTIFIER)):
+    class Servicer(getattr(service_pb2_grpc, SERVICER_IDENTIFIER)):
         pass
 
     server = grpc.server(
         futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-    getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(), server)
+    getattr(service_pb2_grpc, ADD_SERVICER_TO_SERVER_IDENTIFIER)(Servicer(),
+                                                                 server)
     port = server.add_insecure_port('[::]:0')
     server.start()
     channel = grpc.insecure_channel('localhost:{}'.format(port))
-    stub = getattr(service_pb2, STUB_IDENTIFIER)(channel)
+    stub = getattr(service_pb2_grpc, STUB_IDENTIFIER)(channel)
     return _Service(None, server, stub)
 
 
@@ -223,10 +225,11 @@ class PythonPluginTest(unittest.TestCase):
 
     def testImportAttributes(self):
         # check that we can access the generated module and its members.
-        self.assertIsNotNone(getattr(service_pb2, STUB_IDENTIFIER, None))
-        self.assertIsNotNone(getattr(service_pb2, SERVICER_IDENTIFIER, None))
+        self.assertIsNotNone(getattr(service_pb2_grpc, STUB_IDENTIFIER, None))
         self.assertIsNotNone(
-            getattr(service_pb2, ADD_SERVICER_TO_SERVER_IDENTIFIER, None))
+            getattr(service_pb2_grpc, SERVICER_IDENTIFIER, None))
+        self.assertIsNotNone(
+            getattr(service_pb2_grpc, ADD_SERVICER_TO_SERVER_IDENTIFIER, None))
 
     def testUpDown(self):
         service = _CreateService()

src/python/grpcio_tests/tests/protoc_plugin/_split_definitions_test.py

@@ -12,22 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import collections
+import abc
 from concurrent import futures
 import contextlib
-import distutils.spawn
-import errno
 import importlib
 import os
-import os.path
+from os import path
 import pkgutil
+import platform
 import shutil
-import subprocess
 import sys
 import tempfile
-import threading
 import unittest
-import platform
+
+import six
 
 import grpc
 from grpc_tools import protoc
@@ -37,292 +35,285 @@ _MESSAGES_IMPORT = b'import "messages.proto";'
 _SPLIT_NAMESPACE = b'package grpc_protoc_plugin.invocation_testing.split;'
 _COMMON_NAMESPACE = b'package grpc_protoc_plugin.invocation_testing;'
 
+_RELATIVE_PROTO_PATH = 'relative_proto_path'
+_RELATIVE_PYTHON_OUT = 'relative_python_out'
+
 
 @contextlib.contextmanager
-def _system_path(path):
+def _system_path(path_insertion):
     old_system_path = sys.path[:]
-    sys.path = sys.path[0:1] + path + sys.path[1:]
+    sys.path = sys.path[0:1] + path_insertion + sys.path[1:]
     yield
     sys.path = old_system_path
 
 
-class DummySplitServicer(object):
+# NOTE(nathaniel): https://twitter.com/exoplaneteer/status/677259364256747520
+# Life lesson "just always default to idempotence" reinforced.
+def _create_directory_tree(root, path_components_sequence):
+    created = set()
+    for path_components in path_components_sequence:
+        thus_far = ''
+        for path_component in path_components:
+            relative_path = path.join(thus_far, path_component)
+            if relative_path not in created:
+                os.makedirs(path.join(root, relative_path))
+                created.add(relative_path)
+            thus_far = path.join(thus_far, path_component)
+
+
+def _massage_proto_content(proto_content, test_name_bytes,
+                           messages_proto_relative_file_name_bytes):
+    package_substitution = (b'package grpc_protoc_plugin.invocation_testing.' +
+                            test_name_bytes + b';')
+    common_namespace_substituted = proto_content.replace(_COMMON_NAMESPACE,
+                                                         package_substitution)
+    split_namespace_substituted = common_namespace_substituted.replace(
+        _SPLIT_NAMESPACE, package_substitution)
+    message_import_replaced = split_namespace_substituted.replace(
+        _MESSAGES_IMPORT,
+        b'import "' + messages_proto_relative_file_name_bytes + b'";')
+    return message_import_replaced
+
+
+def _packagify(directory):
+    for subdirectory, _, _ in os.walk(directory):
+        init_file_name = path.join(subdirectory, '__init__.py')
+        with open(init_file_name, 'wb') as init_file:
+            init_file.write(b'')
 
-    def __init__(self, request_class, response_class):
-        self.request_class = request_class
-        self.response_class = response_class
+
+class _Servicer(object):
+
+    def __init__(self, response_class):
+        self._response_class = response_class
 
     def Call(self, request, context):
-        return self.response_class()
+        return self._response_class()
 
 
-class SeparateTestMixin(object):
+def _protoc(proto_path, python_out, grpc_python_out_flag, grpc_python_out,
+            absolute_proto_file_names):
+    args = [
+        '',
+        '--proto_path={}'.format(proto_path),
+    ]
+    if python_out is not None:
+        args.append('--python_out={}'.format(python_out))
+    if grpc_python_out is not None:
+        args.append('--grpc_python_out={}:{}'.format(grpc_python_out_flag,
+                                                     grpc_python_out))
+    args.extend(absolute_proto_file_names)
+    return protoc.main(args)
 
-    def testImportAttributes(self):
-        with _system_path([self.python_out_directory]):
-            pb2 = importlib.import_module(self.pb2_import)
-        pb2.Request
-        pb2.Response
-        if self.should_find_services_in_pb2:
-            pb2.TestServiceServicer
-        else:
-            with self.assertRaises(AttributeError):
-                pb2.TestServiceServicer
-
-        with _system_path([self.grpc_python_out_directory]):
-            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-        pb2_grpc.TestServiceServicer
-        with self.assertRaises(AttributeError):
-            pb2_grpc.Request
-        with self.assertRaises(AttributeError):
-            pb2_grpc.Response
-
-    def testCall(self):
-        with _system_path([self.python_out_directory]):
-            pb2 = importlib.import_module(self.pb2_import)
-        with _system_path([self.grpc_python_out_directory]):
-            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-        server = grpc.server(
-            futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-        pb2_grpc.add_TestServiceServicer_to_server(
-            DummySplitServicer(pb2.Request, pb2.Response), server)
-        port = server.add_insecure_port('[::]:0')
-        server.start()
-        channel = grpc.insecure_channel('localhost:{}'.format(port))
-        stub = pb2_grpc.TestServiceStub(channel)
-        request = pb2.Request()
-        expected_response = pb2.Response()
-        response = stub.Call(request)
-        self.assertEqual(expected_response, response)
-
-
-class CommonTestMixin(object):
-
-    def testImportAttributes(self):
-        with _system_path([self.python_out_directory]):
-            pb2 = importlib.import_module(self.pb2_import)
-        pb2.Request
-        pb2.Response
-        if self.should_find_services_in_pb2:
-            pb2.TestServiceServicer
-        else:
-            with self.assertRaises(AttributeError):
-                pb2.TestServiceServicer
-
-        with _system_path([self.grpc_python_out_directory]):
-            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-        pb2_grpc.TestServiceServicer
-        with self.assertRaises(AttributeError):
-            pb2_grpc.Request
-        with self.assertRaises(AttributeError):
-            pb2_grpc.Response
-
-    def testCall(self):
-        with _system_path([self.python_out_directory]):
-            pb2 = importlib.import_module(self.pb2_import)
-        with _system_path([self.grpc_python_out_directory]):
-            pb2_grpc = importlib.import_module(self.pb2_grpc_import)
-        server = grpc.server(
-            futures.ThreadPoolExecutor(max_workers=test_constants.POOL_SIZE))
-        pb2_grpc.add_TestServiceServicer_to_server(
-            DummySplitServicer(pb2.Request, pb2.Response), server)
-        port = server.add_insecure_port('[::]:0')
-        server.start()
-        channel = grpc.insecure_channel('localhost:{}'.format(port))
-        stub = pb2_grpc.TestServiceStub(channel)
-        request = pb2.Request()
-        expected_response = pb2.Response()
-        response = stub.Call(request)
-        self.assertEqual(expected_response, response)
-
-
-@unittest.skipIf(platform.python_implementation() == "PyPy",
-                 "Skip test if run with PyPy")
-class SameSeparateTest(unittest.TestCase, SeparateTestMixin):
 
-    def setUp(self):
-        same_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
-        self.directory = tempfile.mkdtemp(suffix='same_separate', dir='.')
-        self.proto_directory = os.path.join(self.directory, 'proto_path')
-        self.python_out_directory = os.path.join(self.directory, 'python_out')
-        self.grpc_python_out_directory = os.path.join(self.directory,
-                                                      'grpc_python_out')
-        os.makedirs(self.proto_directory)
-        os.makedirs(self.python_out_directory)
-        os.makedirs(self.grpc_python_out_directory)
-        same_proto_file = os.path.join(self.proto_directory,
-                                       'same_separate.proto')
-        open(same_proto_file, 'wb').write(
-            same_proto_contents.replace(
-                _COMMON_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.same_separate;'))
-        protoc_result = protoc.main([
-            '',
-            '--proto_path={}'.format(self.proto_directory),
-            '--python_out={}'.format(self.python_out_directory),
-            '--grpc_python_out=grpc_2_0:{}'.format(
-                self.grpc_python_out_directory),
-            same_proto_file,
-        ])
-        if protoc_result != 0:
-            raise Exception("unexpected protoc error")
-        open(os.path.join(self.grpc_python_out_directory, '__init__.py'),
-             'w').write('')
-        open(os.path.join(self.python_out_directory, '__init__.py'),
-             'w').write('')
-        self.pb2_import = 'same_separate_pb2'
-        self.pb2_grpc_import = 'same_separate_pb2_grpc'
-        self.should_find_services_in_pb2 = False
+class _Mid2016ProtocStyle(object):
 
-    def tearDown(self):
-        shutil.rmtree(self.directory)
+    def name(self):
+        return 'Mid2016ProtocStyle'
 
+    def grpc_in_pb2_expected(self):
+        return True
 
-@unittest.skipIf(platform.python_implementation() == "PyPy",
-                 "Skip test if run with PyPy")
-class SameCommonTest(unittest.TestCase, CommonTestMixin):
+    def protoc(self, proto_path, python_out, absolute_proto_file_names):
+        return (_protoc(proto_path, python_out, 'grpc_1_0', python_out,
+                        absolute_proto_file_names),)
 
-    def setUp(self):
-        same_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing', 'same.proto')
-        self.directory = tempfile.mkdtemp(suffix='same_common', dir='.')
-        self.proto_directory = os.path.join(self.directory, 'proto_path')
-        self.python_out_directory = os.path.join(self.directory, 'python_out')
-        self.grpc_python_out_directory = self.python_out_directory
-        os.makedirs(self.proto_directory)
-        os.makedirs(self.python_out_directory)
-        same_proto_file = os.path.join(self.proto_directory,
-                                       'same_common.proto')
-        open(same_proto_file, 'wb').write(
-            same_proto_contents.replace(
-                _COMMON_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.same_common;'))
-
-        protoc_result = protoc.main([
-            '',
-            '--proto_path={}'.format(self.proto_directory),
-            '--python_out={}'.format(self.python_out_directory),
-            '--grpc_python_out={}'.format(self.grpc_python_out_directory),
-            same_proto_file,
-        ])
-        if protoc_result != 0:
-            raise Exception("unexpected protoc error")
-        open(os.path.join(self.python_out_directory, '__init__.py'),
-             'w').write('')
-        self.pb2_import = 'same_common_pb2'
-        self.pb2_grpc_import = 'same_common_pb2_grpc'
-        self.should_find_services_in_pb2 = True
 
-    def tearDown(self):
-        shutil.rmtree(self.directory)
+class _SingleProtocExecutionProtocStyle(object):
 
+    def name(self):
+        return 'SingleProtocExecutionProtocStyle'
 
-@unittest.skipIf(platform.python_implementation() == "PyPy",
-                 "Skip test if run with PyPy")
-class SplitCommonTest(unittest.TestCase, CommonTestMixin):
+    def grpc_in_pb2_expected(self):
+        return False
 
-    def setUp(self):
-        services_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing.split_services',
-            'services.proto')
-        messages_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing.split_messages',
-            'messages.proto')
-        self.directory = tempfile.mkdtemp(suffix='split_common', dir='.')
-        self.proto_directory = os.path.join(self.directory, 'proto_path')
-        self.python_out_directory = os.path.join(self.directory, 'python_out')
-        self.grpc_python_out_directory = self.python_out_directory
-        os.makedirs(self.proto_directory)
-        os.makedirs(self.python_out_directory)
-        services_proto_file = os.path.join(self.proto_directory,
-                                           'split_common_services.proto')
-        messages_proto_file = os.path.join(self.proto_directory,
-                                           'split_common_messages.proto')
-        open(services_proto_file, 'wb').write(
-            services_proto_contents.replace(
-                _MESSAGES_IMPORT, b'import "split_common_messages.proto";')
-            .replace(
-                _SPLIT_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.split_common;'))
-        open(messages_proto_file, 'wb').write(
-            messages_proto_contents.replace(
-                _SPLIT_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.split_common;'))
-        protoc_result = protoc.main([
-            '',
-            '--proto_path={}'.format(self.proto_directory),
-            '--python_out={}'.format(self.python_out_directory),
-            '--grpc_python_out={}'.format(self.grpc_python_out_directory),
-            services_proto_file,
-            messages_proto_file,
-        ])
-        if protoc_result != 0:
-            raise Exception("unexpected protoc error")
-        open(os.path.join(self.python_out_directory, '__init__.py'),
-             'w').write('')
-        self.pb2_import = 'split_common_messages_pb2'
-        self.pb2_grpc_import = 'split_common_services_pb2_grpc'
-        self.should_find_services_in_pb2 = False
+    def protoc(self, proto_path, python_out, absolute_proto_file_names):
+        return (_protoc(proto_path, python_out, 'grpc_2_0', python_out,
+                        absolute_proto_file_names),)
+
+
+class _ProtoBeforeGrpcProtocStyle(object):
+
+    def name(self):
+        return 'ProtoBeforeGrpcProtocStyle'
+
+    def grpc_in_pb2_expected(self):
+        return False
+
+    def protoc(self, proto_path, python_out, absolute_proto_file_names):
+        pb2_protoc_exit_code = _protoc(proto_path, python_out, None, None,
+                                       absolute_proto_file_names)
+        pb2_grpc_protoc_exit_code = _protoc(
+            proto_path, None, 'grpc_2_0', python_out, absolute_proto_file_names)
+        return pb2_protoc_exit_code, pb2_grpc_protoc_exit_code,
 
-    def tearDown(self):
-        shutil.rmtree(self.directory)
 
+class _GrpcBeforeProtoProtocStyle(object):
 
-@unittest.skipIf(platform.python_implementation() == "PyPy",
-                 "Skip test if run with PyPy")
-class SplitSeparateTest(unittest.TestCase, SeparateTestMixin):
+    def name(self):
+        return 'GrpcBeforeProtoProtocStyle'
+
+    def grpc_in_pb2_expected(self):
+        return False
+
+    def protoc(self, proto_path, python_out, absolute_proto_file_names):
+        pb2_grpc_protoc_exit_code = _protoc(
+            proto_path, None, 'grpc_2_0', python_out, absolute_proto_file_names)
+        pb2_protoc_exit_code = _protoc(proto_path, python_out, None, None,
+                                       absolute_proto_file_names)
+        return pb2_grpc_protoc_exit_code, pb2_protoc_exit_code,
+
+
+_PROTOC_STYLES = (_Mid2016ProtocStyle(), _SingleProtocExecutionProtocStyle(),
+                  _ProtoBeforeGrpcProtocStyle(), _GrpcBeforeProtoProtocStyle(),)
+
+
+@unittest.skipIf(platform.python_implementation() == 'PyPy',
+                 'Skip test if run with PyPy!')
+class _Test(six.with_metaclass(abc.ABCMeta, unittest.TestCase)):
 
     def setUp(self):
-        services_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing.split_services',
-            'services.proto')
-        messages_proto_contents = pkgutil.get_data(
-            'tests.protoc_plugin.protos.invocation_testing.split_messages',
-            'messages.proto')
-        self.directory = tempfile.mkdtemp(suffix='split_separate', dir='.')
-        self.proto_directory = os.path.join(self.directory, 'proto_path')
-        self.python_out_directory = os.path.join(self.directory, 'python_out')
-        self.grpc_python_out_directory = os.path.join(self.directory,
-                                                      'grpc_python_out')
-        os.makedirs(self.proto_directory)
-        os.makedirs(self.python_out_directory)
-        os.makedirs(self.grpc_python_out_directory)
-        services_proto_file = os.path.join(self.proto_directory,
-                                           'split_separate_services.proto')
-        messages_proto_file = os.path.join(self.proto_directory,
-                                           'split_separate_messages.proto')
-        open(services_proto_file, 'wb').write(
-            services_proto_contents.replace(
-                _MESSAGES_IMPORT, b'import "split_separate_messages.proto";')
-            .replace(
-                _SPLIT_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.split_separate;'
-            ))
-        open(messages_proto_file, 'wb').write(
-            messages_proto_contents.replace(
-                _SPLIT_NAMESPACE,
-                b'package grpc_protoc_plugin.invocation_testing.split_separate;'
-            ))
-        protoc_result = protoc.main([
-            '',
-            '--proto_path={}'.format(self.proto_directory),
-            '--python_out={}'.format(self.python_out_directory),
-            '--grpc_python_out=grpc_2_0:{}'.format(
-                self.grpc_python_out_directory),
-            services_proto_file,
-            messages_proto_file,
-        ])
-        if protoc_result != 0:
-            raise Exception("unexpected protoc error")
-        open(os.path.join(self.python_out_directory, '__init__.py'),
-             'w').write('')
-        self.pb2_import = 'split_separate_messages_pb2'
-        self.pb2_grpc_import = 'split_separate_services_pb2_grpc'
-        self.should_find_services_in_pb2 = False
+        self._directory = tempfile.mkdtemp(suffix=self.NAME, dir='.')
+        self._proto_path = path.join(self._directory, _RELATIVE_PROTO_PATH)
+        self._python_out = path.join(self._directory, _RELATIVE_PYTHON_OUT)
+
+        os.makedirs(self._proto_path)
+        os.makedirs(self._python_out)
+
+        proto_directories_and_names = {
+            (self.MESSAGES_PROTO_RELATIVE_DIRECTORY_NAMES,
+             self.MESSAGES_PROTO_FILE_NAME,),
+            (self.SERVICES_PROTO_RELATIVE_DIRECTORY_NAMES,
+             self.SERVICES_PROTO_FILE_NAME,),
+        }
+        messages_proto_relative_file_name_forward_slashes = '/'.join(
+            self.MESSAGES_PROTO_RELATIVE_DIRECTORY_NAMES + (
+                self.MESSAGES_PROTO_FILE_NAME,))
+        _create_directory_tree(self._proto_path, (
+            relative_proto_directory_names
+            for relative_proto_directory_names, _ in proto_directories_and_names
+        ))
+        self._absolute_proto_file_names = set()
+        for relative_directory_names, file_name in proto_directories_and_names:
+            absolute_proto_file_name = path.join(
+                self._proto_path, *relative_directory_names + (file_name,))
+            raw_proto_content = pkgutil.get_data(
+                'tests.protoc_plugin.protos.invocation_testing',
+                path.join(*relative_directory_names + (file_name,)))
+            massaged_proto_content = _massage_proto_content(
+                raw_proto_content,
+                self.NAME.encode(),
+                messages_proto_relative_file_name_forward_slashes.encode())
+            with open(absolute_proto_file_name, 'wb') as proto_file:
+                proto_file.write(massaged_proto_content)
+            self._absolute_proto_file_names.add(absolute_proto_file_name)
 
     def tearDown(self):
-        shutil.rmtree(self.directory)
+        shutil.rmtree(self._directory)
+
+    def _protoc(self):
+        protoc_exit_codes = self.PROTOC_STYLE.protoc(
+            self._proto_path, self._python_out, self._absolute_proto_file_names)
+        for protoc_exit_code in protoc_exit_codes:
+            self.assertEqual(0, protoc_exit_code)
+
+        _packagify(self._python_out)
+
+        generated_modules = {}
+        expected_generated_full_module_names = {
+            self.EXPECTED_MESSAGES_PB2,
+            self.EXPECTED_SERVICES_PB2,
+            self.EXPECTED_SERVICES_PB2_GRPC,
+        }
+        with _system_path([self._python_out]):
+            for full_module_name in expected_generated_full_module_names:
+                module = importlib.import_module(full_module_name)
+                generated_modules[full_module_name] = module
+
+        self._messages_pb2 = generated_modules[self.EXPECTED_MESSAGES_PB2]
+        self._services_pb2 = generated_modules[self.EXPECTED_SERVICES_PB2]
+        self._services_pb2_grpc = generated_modules[
+            self.EXPECTED_SERVICES_PB2_GRPC]
+
+    def _services_modules(self):
+        if self.PROTOC_STYLE.grpc_in_pb2_expected():
+            return self._services_pb2, self._services_pb2_grpc,
+        else:
+            return self._services_pb2_grpc,
+
+    def test_imported_attributes(self):
+        self._protoc()
+
+        self._messages_pb2.Request
+        self._messages_pb2.Response
+        self._services_pb2.DESCRIPTOR.services_by_name['TestService']
+        for services_module in self._services_modules():
+            services_module.TestServiceStub
+            services_module.TestServiceServicer
+            services_module.add_TestServiceServicer_to_server
+
+    def test_call(self):
+        self._protoc()
+
+        for services_module in self._services_modules():
+            server = grpc.server(
+                futures.ThreadPoolExecutor(
+                    max_workers=test_constants.POOL_SIZE))
+            services_module.add_TestServiceServicer_to_server(
+                _Servicer(self._messages_pb2.Response), server)
+            port = server.add_insecure_port('[::]:0')
+            server.start()
+            channel = grpc.insecure_channel('localhost:{}'.format(port))
+            stub = services_module.TestServiceStub(channel)
+            response = stub.Call(self._messages_pb2.Request())
+            self.assertEqual(self._messages_pb2.Response(), response)
+
+
+def _create_test_case_class(split_proto, protoc_style):
+    attributes = {}
+
+    name = '{}{}'.format('SplitProto' if split_proto else 'SameProto',
+                         protoc_style.name())
+    attributes['NAME'] = name
+
+    if split_proto:
+        attributes['MESSAGES_PROTO_RELATIVE_DIRECTORY_NAMES'] = (
+            'split_messages', 'sub',)
+        attributes['MESSAGES_PROTO_FILE_NAME'] = 'messages.proto'
+        attributes['SERVICES_PROTO_RELATIVE_DIRECTORY_NAMES'] = (
+            'split_services',)
+        attributes['SERVICES_PROTO_FILE_NAME'] = 'services.proto'
+        attributes['EXPECTED_MESSAGES_PB2'] = 'split_messages.sub.messages_pb2'
+        attributes['EXPECTED_SERVICES_PB2'] = 'split_services.services_pb2'
+        attributes['EXPECTED_SERVICES_PB2_GRPC'] = (
+            'split_services.services_pb2_grpc')
+    else:
+        attributes['MESSAGES_PROTO_RELATIVE_DIRECTORY_NAMES'] = ()
+        attributes['MESSAGES_PROTO_FILE_NAME'] = 'same.proto'
+        attributes['SERVICES_PROTO_RELATIVE_DIRECTORY_NAMES'] = ()
+        attributes['SERVICES_PROTO_FILE_NAME'] = 'same.proto'
+        attributes['EXPECTED_MESSAGES_PB2'] = 'same_pb2'
+        attributes['EXPECTED_SERVICES_PB2'] = 'same_pb2'
+        attributes['EXPECTED_SERVICES_PB2_GRPC'] = 'same_pb2_grpc'
+
+    attributes['PROTOC_STYLE'] = protoc_style
+
+    attributes['__module__'] = _Test.__module__
+
+    return type('{}Test'.format(name), (_Test,), attributes)
+
+
+def _create_test_case_classes():
+    for split_proto in (False, True,):
+        for protoc_style in _PROTOC_STYLES:
+            yield _create_test_case_class(split_proto, protoc_style)
+
+
+def load_tests(loader, tests, pattern):
+    tests = tuple(
+        loader.loadTestsFromTestCase(test_case_class)
+        for test_case_class in _create_test_case_classes())
+    return unittest.TestSuite(tests=tests)
 
 
 if __name__ == '__main__':

src/python/grpcio_tests/tests/qps/benchmark_server.py

@@ -13,10 +13,10 @@
 # limitations under the License.
 
 from src.proto.grpc.testing import messages_pb2
-from src.proto.grpc.testing import services_pb2
+from src.proto.grpc.testing import services_pb2_grpc
 
 
-class BenchmarkServer(services_pb2.BenchmarkServiceServicer):
+class BenchmarkServer(services_pb2_grpc.BenchmarkServiceServicer):
     """Synchronous Server implementation for the Benchmark service."""
 
     def UnaryCall(self, request, context):
@@ -29,7 +29,7 @@ class BenchmarkServer(services_pb2.BenchmarkServiceServicer):
             yield messages_pb2.SimpleResponse(payload=payload)
 
 
-class GenericBenchmarkServer(services_pb2.BenchmarkServiceServicer):
+class GenericBenchmarkServer(services_pb2_grpc.BenchmarkServiceServicer):
     """Generic Server implementation for the Benchmark service."""
 
     def __init__(self, resp_size):

src/python/grpcio_tests/tests/stress/metrics_server.py

@@ -16,11 +16,12 @@
 import time
 
 from src.proto.grpc.testing import metrics_pb2
+from src.proto.grpc.testing import metrics_pb2_grpc
 
 GAUGE_NAME = 'python_overall_qps'
 
 
-class MetricsServer(metrics_pb2.MetricsServiceServicer):
+class MetricsServer(metrics_pb2_grpc.MetricsServiceServicer):
 
     def __init__(self, histogram):
         self._start_time = time.time()

src/python/grpcio_tests/tests/tests.json

@@ -1,12 +1,17 @@
 [
+  "_sanity._sanity_test.SanityTest",
   "health_check._health_servicer_test.HealthServicerTest",
   "interop._insecure_intraop_test.InsecureIntraopTest",
   "interop._secure_intraop_test.SecureIntraopTest",
   "protoc_plugin._python_plugin_test.PythonPluginTest",
-  "protoc_plugin._split_definitions_test.SameCommonTest",
-  "protoc_plugin._split_definitions_test.SameSeparateTest",
-  "protoc_plugin._split_definitions_test.SplitCommonTest",
-  "protoc_plugin._split_definitions_test.SplitSeparateTest",
+  "protoc_plugin._split_definitions_test.SameProtoGrpcBeforeProtoProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SameProtoMid2016ProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SameProtoProtoBeforeGrpcProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SameProtoSingleProtocExecutionProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SplitProtoGrpcBeforeProtoProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SplitProtoMid2016ProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SplitProtoProtoBeforeGrpcProtocStyleTest",
+  "protoc_plugin._split_definitions_test.SplitProtoSingleProtocExecutionProtocStyleTest",
   "protoc_plugin.beta_python_plugin_test.PythonPluginTest",
   "reflection._reflection_servicer_test.ReflectionServicerTest",
   "testing._client_test.ClientTest",
@@ -41,7 +46,6 @@
   "unit._reconnect_test.ReconnectTest",
   "unit._resource_exhausted_test.ResourceExhaustedTest",
   "unit._rpc_test.RPCTest",
-  "unit._sanity._sanity_test.Sanity",
   "unit._thread_cleanup_test.CleanupThreadTest",
   "unit.beta._beta_features_test.BetaFeaturesTest",
   "unit.beta._beta_features_test.ContextManagementAndLifecycleTest",

test/cpp/codegen/compiler_test_golden

@@ -65,6 +65,9 @@ class ServiceA final {
     std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
       return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq));
     }
+    std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodA1Raw(context, request, cq));
+    }
     // MethodA1 trailing comment 1
     // MethodA2 detached leading comment 1
     //
@@ -76,6 +79,9 @@ class ServiceA final {
     std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag));
     }
+    std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>> PrepareAsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>>(PrepareAsyncMethodA2Raw(context, response, cq));
+    }
     // MethodA2 trailing comment 1
     // Method A3 leading comment 1
     std::unique_ptr< ::grpc::ClientReaderInterface< ::grpc::testing::Response>> MethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request) {
@@ -84,6 +90,9 @@ class ServiceA final {
     std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>> AsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>>(AsyncMethodA3Raw(context, request, cq, tag));
     }
+    std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodA3Raw(context, request, cq));
+    }
     // Method A3 trailing comment 1
     // Method A4 leading comment 1
     std::unique_ptr< ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> MethodA4(::grpc::ClientContext* context) {
@@ -92,15 +101,22 @@ class ServiceA final {
     std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>>(AsyncMethodA4Raw(context, cq, tag));
     }
+    std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> PrepareAsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>>(PrepareAsyncMethodA4Raw(context, cq));
+    }
     // Method A4 trailing comment 1
   private:
     virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
+    virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
     virtual ::grpc::ClientWriterInterface< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) = 0;
     virtual ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) = 0;
+    virtual ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>* PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) = 0;
     virtual ::grpc::ClientReaderInterface< ::grpc::testing::Response>* MethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request) = 0;
     virtual ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>* AsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) = 0;
+    virtual ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
     virtual ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4Raw(::grpc::ClientContext* context) = 0;
     virtual ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* AsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) = 0;
+    virtual ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) = 0;
   };
   class Stub final : public StubInterface {
    public:
@@ -109,34 +125,50 @@ class ServiceA final {
     std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
       return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq));
     }
+    std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> PrepareAsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(PrepareAsyncMethodA1Raw(context, request, cq));
+    }
     std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>> MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response) {
       return std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>>(MethodA2Raw(context, response));
     }
     std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag));
     }
+    std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>> PrepareAsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>>(PrepareAsyncMethodA2Raw(context, response, cq));
+    }
     std::unique_ptr< ::grpc::ClientReader< ::grpc::testing::Response>> MethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request) {
       return std::unique_ptr< ::grpc::ClientReader< ::grpc::testing::Response>>(MethodA3Raw(context, request));
     }
     std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>> AsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>>(AsyncMethodA3Raw(context, request, cq, tag));
     }
+    std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>> PrepareAsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>>(PrepareAsyncMethodA3Raw(context, request, cq));
+    }
     std::unique_ptr< ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> MethodA4(::grpc::ClientContext* context) {
       return std::unique_ptr< ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(MethodA4Raw(context));
     }
     std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) {
       return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(AsyncMethodA4Raw(context, cq, tag));
     }
+    std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> PrepareAsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(PrepareAsyncMethodA4Raw(context, cq));
+    }
 
    private:
     std::shared_ptr< ::grpc::ChannelInterface> channel_;
     ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
+    ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
     ::grpc::ClientWriter< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) override;
     ::grpc::ClientAsyncWriter< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) override;
+    ::grpc::ClientAsyncWriter< ::grpc::testing::Request>* PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) override;
     ::grpc::ClientReader< ::grpc::testing::Response>* MethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request) override;
     ::grpc::ClientAsyncReader< ::grpc::testing::Response>* AsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) override;
+    ::grpc::ClientAsyncReader< ::grpc::testing::Response>* PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
     ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4Raw(::grpc::ClientContext* context) override;
     ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* AsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) override;
+    ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) override;
     const ::grpc::RpcMethod rpcmethod_MethodA1_;
     const ::grpc::RpcMethod rpcmethod_MethodA2_;
     const ::grpc::RpcMethod rpcmethod_MethodA3_;
@@ -372,9 +404,13 @@ class ServiceB final {
     std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
       return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq));
     }
+    std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodB1Raw(context, request, cq));
+    }
     // MethodB1 trailing comment 1
   private:
     virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
+    virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0;
   };
   class Stub final : public StubInterface {
    public:
@@ -383,10 +419,14 @@ class ServiceB final {
     std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
       return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq));
     }
+    std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> PrepareAsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(PrepareAsyncMethodB1Raw(context, request, cq));
+    }
 
    private:
     std::shared_ptr< ::grpc::ChannelInterface> channel_;
     ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
+    ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
     const ::grpc::RpcMethod rpcmethod_MethodB1_;
   };
   static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());

test/cpp/codegen/compiler_test_mock_golden

@@ -15,18 +15,23 @@ class MockServiceAStub : public ServiceA::StubInterface {
  public:
   MOCK_METHOD3(MethodA1, ::grpc::Status(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response));
   MOCK_METHOD3(AsyncMethodA1Raw, ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq));
+  MOCK_METHOD3(PrepareAsyncMethodA1Raw, ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq));
   MOCK_METHOD2(MethodA2Raw, ::grpc::ClientWriterInterface< ::grpc::testing::Request>*(::grpc::ClientContext* context, ::grpc::testing::Response* response));
   MOCK_METHOD4(AsyncMethodA2Raw, ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>*(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag));
+  MOCK_METHOD3(PrepareAsyncMethodA2Raw, ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>*(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq));
   MOCK_METHOD2(MethodA3Raw, ::grpc::ClientReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request));
   MOCK_METHOD4(AsyncMethodA3Raw, ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag));
+  MOCK_METHOD3(PrepareAsyncMethodA3Raw, ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq));
   MOCK_METHOD1(MethodA4Raw, ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>*(::grpc::ClientContext* context));
   MOCK_METHOD3(AsyncMethodA4Raw, ::grpc::ClientAsyncReaderWriterInterface<::grpc::testing::Request, ::grpc::testing::Response>*(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag));
+  MOCK_METHOD2(PrepareAsyncMethodA4Raw, ::grpc::ClientAsyncReaderWriterInterface<::grpc::testing::Request, ::grpc::testing::Response>*(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq));
 };
 
 class MockServiceBStub : public ServiceB::StubInterface {
  public:
   MOCK_METHOD3(MethodB1, ::grpc::Status(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response));
   MOCK_METHOD3(AsyncMethodB1Raw, ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq));
+  MOCK_METHOD3(PrepareAsyncMethodB1Raw, ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>*(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq));
 };
 
 } // namespace grpc

test/cpp/qps/client_async.cc

@@ -56,11 +56,6 @@ class ClientRpcContext {
   }
 
   virtual void Start(CompletionQueue* cq, const ClientConfig& config) = 0;
-  void lock() { mu_.lock(); }
-  void unlock() { mu_.unlock(); }
-
- private:
-  std::mutex mu_;
 };
 
 template <class RequestType, class ResponseType>
@@ -73,7 +68,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
           std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
               BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
               CompletionQueue*)>
-          start_req,
+          prepare_req,
       std::function<void(grpc::Status, ResponseType*, HistogramEntry*)> on_done)
       : context_(),
         stub_(stub),
@@ -83,7 +78,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
         next_state_(State::READY),
         callback_(on_done),
         next_issue_(next_issue),
-        start_req_(start_req) {}
+        prepare_req_(prepare_req) {}
   ~ClientRpcContextUnaryImpl() override {}
   void Start(CompletionQueue* cq, const ClientConfig& config) override {
     StartInternal(cq);
@@ -92,7 +87,8 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
     switch (next_state_) {
       case State::READY:
         start_ = UsageTimer::Now();
-        response_reader_ = start_req_(stub_, &context_, req_, cq_);
+        response_reader_ = prepare_req_(stub_, &context_, req_, cq_);
+        response_reader_->StartCall();
         next_state_ = State::RESP_DONE;
         response_reader_->Finish(&response_, &status_,
                                  ClientRpcContext::tag(this));
@@ -111,8 +107,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
   }
   void StartNewClone(CompletionQueue* cq) override {
     auto* clone = new ClientRpcContextUnaryImpl(stub_, req_, next_issue_,
-                                                start_req_, callback_);
-    std::lock_guard<ClientRpcContext> lclone(*clone);
+                                                prepare_req_, callback_);
     clone->StartInternal(cq);
   }
 
@@ -130,7 +125,7 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext {
   std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>(
       BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
       CompletionQueue*)>
-      start_req_;
+      prepare_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>
@@ -252,29 +247,13 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
       // this thread isn't supposed to shut down
       std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
       if (shutdown_state_[thread_idx]->shutdown) {
-        // We want to delete the context. However, it is possible that
-        // another thread that just initiated an action on this
-        // context still has its lock even though the action on the
-        // context has completed. To delay for that, just grab the
-        // lock for serialization. Take a new scope.
-        { std::lock_guard<ClientRpcContext> lctx(*ctx); }
         delete ctx;
         return true;
       }
-      bool del = false;
-
-      // Create a new scope for a lock_guard'ed region
-      {
-        std::lock_guard<ClientRpcContext> lctx(*ctx);
-        if (!ctx->RunNextState(ok, entry)) {
-          // The RPC and callback are done, so clone the ctx
-          // and kickstart the new one
-          ctx->StartNewClone(cli_cqs_[cq_[thread_idx]].get());
-          // set the old version to delete
-          del = true;
-        }
-      }
-      if (del) {
+      if (!ctx->RunNextState(ok, entry)) {
+        // The RPC and callback are done, so clone the ctx
+        // and kickstart the new one
+        ctx->StartNewClone(cli_cqs_[cq_[thread_idx]].get());
         delete ctx;
       }
       return true;
@@ -311,15 +290,15 @@ class AsyncUnaryClient final
     entry->set_status(s.error_code());
   }
   static std::unique_ptr<grpc::ClientAsyncResponseReader<SimpleResponse>>
-  StartReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
-           const SimpleRequest& request, CompletionQueue* cq) {
-    return stub->AsyncUnaryCall(ctx, request, cq);
+  PrepareReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
+             const SimpleRequest& request, CompletionQueue* cq) {
+    return stub->PrepareAsyncUnaryCall(ctx, request, cq);
   };
   static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
                                     std::function<gpr_timespec()> next_issue,
                                     const SimpleRequest& req) {
     return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
-        stub, req, next_issue, AsyncUnaryClient::StartReq,
+        stub, req, next_issue, AsyncUnaryClient::PrepareReq,
         AsyncUnaryClient::CheckDone);
   }
 };
@@ -332,9 +311,8 @@ class ClientRpcContextStreamingPingPongImpl : public ClientRpcContext {
       std::function<gpr_timespec()> next_issue,
       std::function<std::unique_ptr<
           grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
-          BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*,
-          void*)>
-          start_req,
+          BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*)>
+          prepare_req,
       std::function<void(grpc::Status, ResponseType*)> on_done)
       : context_(),
         stub_(stub),
@@ -344,7 +322,7 @@ class ClientRpcContextStreamingPingPongImpl : public ClientRpcContext {
         next_state_(State::INVALID),
         callback_(on_done),
         next_issue_(next_issue),
-        start_req_(start_req) {}
+        prepare_req_(prepare_req) {}
   ~ClientRpcContextStreamingPingPongImpl() override {}
   void Start(CompletionQueue* cq, const ClientConfig& config) override {
     StartInternal(cq, config.messages_per_stream());
@@ -407,8 +385,7 @@ class ClientRpcContextStreamingPingPongImpl : public ClientRpcContext {
   }
   void StartNewClone(CompletionQueue* cq) override {
     auto* clone = new ClientRpcContextStreamingPingPongImpl(
-        stub_, req_, next_issue_, start_req_, callback_);
-    std::lock_guard<ClientRpcContext> lclone(*clone);
+        stub_, req_, next_issue_, prepare_req_, callback_);
     clone->StartInternal(cq, messages_per_stream_);
   }
 
@@ -432,10 +409,10 @@ class ClientRpcContextStreamingPingPongImpl : public ClientRpcContext {
   State next_state_;
   std::function<void(grpc::Status, ResponseType*)> callback_;
   std::function<gpr_timespec()> next_issue_;
-  std::function<std::unique_ptr<
-      grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
-      BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*, void*)>
-      start_req_;
+  std::function<
+      std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>(
+          BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*)>
+      prepare_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>
@@ -449,8 +426,9 @@ class ClientRpcContextStreamingPingPongImpl : public ClientRpcContext {
     cq_ = cq;
     messages_per_stream_ = messages_per_stream;
     messages_issued_ = 0;
+    stream_ = prepare_req_(stub_, &context_, cq);
     next_state_ = State::STREAM_IDLE;
-    stream_ = start_req_(stub_, &context_, cq, ClientRpcContext::tag(this));
+    stream_->StartCall(ClientRpcContext::tag(this));
   }
 };
 
@@ -469,9 +447,9 @@ class AsyncStreamingPingPongClient final
   static void CheckDone(grpc::Status s, SimpleResponse* response) {}
   static std::unique_ptr<
       grpc::ClientAsyncReaderWriter<SimpleRequest, SimpleResponse>>
-  StartReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
-           CompletionQueue* cq, void* tag) {
-    auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
+  PrepareReq(BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
+             CompletionQueue* cq) {
+    auto stream = stub->PrepareAsyncStreamingCall(ctx, cq);
     return stream;
   };
   static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
@@ -479,7 +457,7 @@ class AsyncStreamingPingPongClient final
                                     const SimpleRequest& req) {
     return new ClientRpcContextStreamingPingPongImpl<SimpleRequest,
                                                      SimpleResponse>(
-        stub, req, next_issue, AsyncStreamingPingPongClient::StartReq,
+        stub, req, next_issue, AsyncStreamingPingPongClient::PrepareReq,
         AsyncStreamingPingPongClient::CheckDone);
   }
 };
@@ -492,8 +470,8 @@ class ClientRpcContextStreamingFromClientImpl : public ClientRpcContext {
       std::function<gpr_timespec()> next_issue,
       std::function<std::unique_ptr<grpc::ClientAsyncWriter<RequestType>>(
           BenchmarkService::Stub*, grpc::ClientContext*, ResponseType*,
-          CompletionQueue*, void*)>
-          start_req,
+          CompletionQueue*)>
+          prepare_req,
       std::function<void(grpc::Status, ResponseType*)> on_done)
       : context_(),
         stub_(stub),
@@ -503,7 +481,7 @@ class ClientRpcContextStreamingFromClientImpl : public ClientRpcContext {
         next_state_(State::INVALID),
         callback_(on_done),
         next_issue_(next_issue),
-        start_req_(start_req) {}
+        prepare_req_(prepare_req) {}
   ~ClientRpcContextStreamingFromClientImpl() override {}
   void Start(CompletionQueue* cq, const ClientConfig& config) override {
     StartInternal(cq);
@@ -546,8 +524,7 @@ class ClientRpcContextStreamingFromClientImpl : public ClientRpcContext {
   }
   void StartNewClone(CompletionQueue* cq) override {
     auto* clone = new ClientRpcContextStreamingFromClientImpl(
-        stub_, req_, next_issue_, start_req_, callback_);
-    std::lock_guard<ClientRpcContext> lclone(*clone);
+        stub_, req_, next_issue_, prepare_req_, callback_);
     clone->StartInternal(cq);
   }
 
@@ -570,17 +547,17 @@ class ClientRpcContextStreamingFromClientImpl : public ClientRpcContext {
   std::function<gpr_timespec()> next_issue_;
   std::function<std::unique_ptr<grpc::ClientAsyncWriter<RequestType>>(
       BenchmarkService::Stub*, grpc::ClientContext*, ResponseType*,
-      CompletionQueue*, void*)>
-      start_req_;
+      CompletionQueue*)>
+      prepare_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::ClientAsyncWriter<RequestType>> stream_;
 
   void StartInternal(CompletionQueue* cq) {
     cq_ = cq;
-    stream_ = start_req_(stub_, &context_, &response_, cq,
-                         ClientRpcContext::tag(this));
+    stream_ = prepare_req_(stub_, &context_, &response_, cq);
     next_state_ = State::STREAM_IDLE;
+    stream_->StartCall(ClientRpcContext::tag(this));
   }
 };
 
@@ -597,10 +574,10 @@ class AsyncStreamingFromClientClient final
 
  private:
   static void CheckDone(grpc::Status s, SimpleResponse* response) {}
-  static std::unique_ptr<grpc::ClientAsyncWriter<SimpleRequest>> StartReq(
+  static std::unique_ptr<grpc::ClientAsyncWriter<SimpleRequest>> PrepareReq(
       BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
-      SimpleResponse* resp, CompletionQueue* cq, void* tag) {
-    auto stream = stub->AsyncStreamingFromClient(ctx, resp, cq, tag);
+      SimpleResponse* resp, CompletionQueue* cq) {
+    auto stream = stub->PrepareAsyncStreamingFromClient(ctx, resp, cq);
     return stream;
   };
   static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
@@ -608,7 +585,7 @@ class AsyncStreamingFromClientClient final
                                     const SimpleRequest& req) {
     return new ClientRpcContextStreamingFromClientImpl<SimpleRequest,
                                                        SimpleResponse>(
-        stub, req, next_issue, AsyncStreamingFromClientClient::StartReq,
+        stub, req, next_issue, AsyncStreamingFromClientClient::PrepareReq,
         AsyncStreamingFromClientClient::CheckDone);
   }
 };
@@ -621,8 +598,8 @@ class ClientRpcContextStreamingFromServerImpl : public ClientRpcContext {
       std::function<gpr_timespec()> next_issue,
       std::function<std::unique_ptr<grpc::ClientAsyncReader<ResponseType>>(
           BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
-          CompletionQueue*, void*)>
-          start_req,
+          CompletionQueue*)>
+          prepare_req,
       std::function<void(grpc::Status, ResponseType*)> on_done)
       : context_(),
         stub_(stub),
@@ -632,7 +609,7 @@ class ClientRpcContextStreamingFromServerImpl : public ClientRpcContext {
         next_state_(State::INVALID),
         callback_(on_done),
         next_issue_(next_issue),
-        start_req_(start_req) {}
+        prepare_req_(prepare_req) {}
   ~ClientRpcContextStreamingFromServerImpl() override {}
   void Start(CompletionQueue* cq, const ClientConfig& config) override {
     StartInternal(cq);
@@ -664,8 +641,7 @@ class ClientRpcContextStreamingFromServerImpl : public ClientRpcContext {
   }
   void StartNewClone(CompletionQueue* cq) override {
     auto* clone = new ClientRpcContextStreamingFromServerImpl(
-        stub_, req_, next_issue_, start_req_, callback_);
-    std::lock_guard<ClientRpcContext> lclone(*clone);
+        stub_, req_, next_issue_, prepare_req_, callback_);
     clone->StartInternal(cq);
   }
 
@@ -682,8 +658,8 @@ class ClientRpcContextStreamingFromServerImpl : public ClientRpcContext {
   std::function<gpr_timespec()> next_issue_;
   std::function<std::unique_ptr<grpc::ClientAsyncReader<ResponseType>>(
       BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&,
-      CompletionQueue*, void*)>
-      start_req_;
+      CompletionQueue*)>
+      prepare_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::ClientAsyncReader<ResponseType>> stream_;
@@ -691,9 +667,9 @@ class ClientRpcContextStreamingFromServerImpl : public ClientRpcContext {
   void StartInternal(CompletionQueue* cq) {
     // TODO(vjpai): Add support to rate-pace this
     cq_ = cq;
+    stream_ = prepare_req_(stub_, &context_, req_, cq);
     next_state_ = State::STREAM_IDLE;
-    stream_ =
-        start_req_(stub_, &context_, req_, cq, ClientRpcContext::tag(this));
+    stream_->StartCall(ClientRpcContext::tag(this));
   }
 };
 
@@ -710,10 +686,10 @@ class AsyncStreamingFromServerClient final
 
  private:
   static void CheckDone(grpc::Status s, SimpleResponse* response) {}
-  static std::unique_ptr<grpc::ClientAsyncReader<SimpleResponse>> StartReq(
+  static std::unique_ptr<grpc::ClientAsyncReader<SimpleResponse>> PrepareReq(
       BenchmarkService::Stub* stub, grpc::ClientContext* ctx,
-      const SimpleRequest& req, CompletionQueue* cq, void* tag) {
-    auto stream = stub->AsyncStreamingFromServer(ctx, req, cq, tag);
+      const SimpleRequest& req, CompletionQueue* cq) {
+    auto stream = stub->PrepareAsyncStreamingFromServer(ctx, req, cq);
     return stream;
   };
   static ClientRpcContext* SetupCtx(BenchmarkService::Stub* stub,
@@ -721,7 +697,7 @@ class AsyncStreamingFromServerClient final
                                     const SimpleRequest& req) {
     return new ClientRpcContextStreamingFromServerImpl<SimpleRequest,
                                                        SimpleResponse>(
-        stub, req, next_issue, AsyncStreamingFromServerClient::StartReq,
+        stub, req, next_issue, AsyncStreamingFromServerClient::PrepareReq,
         AsyncStreamingFromServerClient::CheckDone);
   }
 };
@@ -733,8 +709,8 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
       std::function<gpr_timespec()> next_issue,
       std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
           grpc::GenericStub*, grpc::ClientContext*,
-          const grpc::string& method_name, CompletionQueue*, void*)>
-          start_req,
+          const grpc::string& method_name, CompletionQueue*)>
+          prepare_req,
       std::function<void(grpc::Status, ByteBuffer*)> on_done)
       : context_(),
         stub_(stub),
@@ -744,7 +720,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
         next_state_(State::INVALID),
         callback_(on_done),
         next_issue_(next_issue),
-        start_req_(start_req) {}
+        prepare_req_(prepare_req) {}
   ~ClientRpcContextGenericStreamingImpl() override {}
   void Start(CompletionQueue* cq, const ClientConfig& config) override {
     StartInternal(cq, config.messages_per_stream());
@@ -807,8 +783,7 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
   }
   void StartNewClone(CompletionQueue* cq) override {
     auto* clone = new ClientRpcContextGenericStreamingImpl(
-        stub_, req_, next_issue_, start_req_, callback_);
-    std::lock_guard<ClientRpcContext> lclone(*clone);
+        stub_, req_, next_issue_, prepare_req_, callback_);
     clone->StartInternal(cq, messages_per_stream_);
   }
 
@@ -834,8 +809,8 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
   std::function<gpr_timespec()> next_issue_;
   std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>(
       grpc::GenericStub*, grpc::ClientContext*, const grpc::string&,
-      CompletionQueue*, void*)>
-      start_req_;
+      CompletionQueue*)>
+      prepare_req_;
   grpc::Status status_;
   double start_;
   std::unique_ptr<grpc::GenericClientAsyncReaderWriter> stream_;
@@ -850,9 +825,9 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext {
         "/grpc.testing.BenchmarkService/StreamingCall");
     messages_per_stream_ = messages_per_stream;
     messages_issued_ = 0;
+    stream_ = prepare_req_(stub_, &context_, kMethodName, cq);
     next_state_ = State::STREAM_IDLE;
-    stream_ = start_req_(stub_, &context_, kMethodName, cq,
-                         ClientRpcContext::tag(this));
+    stream_->StartCall(ClientRpcContext::tag(this));
   }
 };
 
@@ -874,17 +849,17 @@ class GenericAsyncStreamingClient final
 
  private:
   static void CheckDone(grpc::Status s, ByteBuffer* response) {}
-  static std::unique_ptr<grpc::GenericClientAsyncReaderWriter> StartReq(
+  static std::unique_ptr<grpc::GenericClientAsyncReaderWriter> PrepareReq(
       grpc::GenericStub* stub, grpc::ClientContext* ctx,
-      const grpc::string& method_name, CompletionQueue* cq, void* tag) {
-    auto stream = stub->Call(ctx, method_name, cq, tag);
+      const grpc::string& method_name, CompletionQueue* cq) {
+    auto stream = stub->PrepareCall(ctx, method_name, cq);
     return stream;
   };
   static ClientRpcContext* SetupCtx(grpc::GenericStub* stub,
                                     std::function<gpr_timespec()> next_issue,
                                     const ByteBuffer& req) {
     return new ClientRpcContextGenericStreamingImpl(
-        stub, req, next_issue, GenericAsyncStreamingClient::StartReq,
+        stub, req, next_issue, GenericAsyncStreamingClient::PrepareReq,
         GenericAsyncStreamingClient::CheckDone);
   }
 };

tools/internal_ci/helper_scripts/prepare_build_linux_perf_rc

@@ -19,4 +19,19 @@
 ulimit -n 32768
 ulimit -c unlimited
 
+# Performance PR testing needs GH API key and PR metadata to comment results
+if [ -n "$KOKORO_GITHUB_PULL_REQUEST_NUMBER" ]; then
+  set +x
+  sudo apt-get install -y jq
+  export ghprbTargetBranch=$(curl -s https://api.github.com/repos/grpc/grpc/pulls/$KOKORO_GITHUB_PULL_REQUEST_NUMBER | jq -r .base.ref)
+
+  gsutil cp gs://grpc-testing-secrets/github_credentials/oauth_token.txt ~/
+  # TODO(matt-kwong): rename this to GITHUB_OAUTH_TOKEN after Jenkins deprecation
+  export JENKINS_OAUTH_TOKEN=$(cat ~/oauth_token.txt)
+  export ghprbPullId=$KOKORO_GITHUB_PULL_REQUEST_NUMBER
+  set -x
+fi
+
+sudo pip install tabulate
+
 git submodule update --init

tools/internal_ci/linux/grpc_microbenchmark_diff.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+# Copyright 2017 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.
+#
+# This script is invoked by Jenkins and runs a diff on the microbenchmarks
+set -ex
+
+# List of benchmarks that provide good signal for analyzing performance changes in pull requests
+BENCHMARKS_TO_RUN="bm_fullstack_unary_ping_pong bm_fullstack_streaming_ping_pong bm_fullstack_streaming_pump bm_closure bm_cq bm_call_create bm_error bm_chttp2_hpack bm_chttp2_transport bm_pollset bm_metadata"
+
+# Enter the gRPC repo root
+cd $(dirname $0)/../../..
+
+source tools/internal_ci/helper_scripts/prepare_build_linux_perf_rc
+
+tools/run_tests/start_port_server.py
+tools/jenkins/run_c_cpp_test.sh tools/profiling/microbenchmarks/bm_diff/bm_main.py \
+  -d origin/$ghprbTargetBranch \
+  -b $BENCHMARKS_TO_RUN || FAILED="true"
+
+# kill port_server.py to prevent the build from hanging
+ps aux | grep port_server\\.py | awk '{print $2}' | xargs kill -9
+
+if [ "$FAILED" != "" ]
+then
+  exit 1
+fi

tools/internal_ci/linux/grpc_run_tests_matrix.sh

@@ -20,4 +20,14 @@ cd $(dirname $0)/../../..
 
 source tools/internal_ci/helper_scripts/prepare_build_linux_rc
 
-tools/run_tests/run_tests_matrix.py $RUN_TESTS_FLAGS
+tools/run_tests/run_tests_matrix.py $RUN_TESTS_FLAGS || FAILED="true"
+
+# Reveal leftover processes that might be left behind by the build
+ps aux | grep -i kbuilder
+
+echo 'Exiting gRPC main test script.'
+
+if [ "$FAILED" != "" ]
+then
+  exit 1
+fi

tools/internal_ci/linux/grpc_trickle_diff.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# Copyright 2017 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.
+#
+# This script is invoked by Jenkins and runs a diff on the microbenchmarks
+set -ex
+
+# List of benchmarks that provide good signal for analyzing performance changes in pull requests
+BENCHMARKS_TO_RUN="cli_transport_stalls_per_iteration cli_stream_stalls_per_iteration svr_transport_stalls_per_iteration svr_stream_stalls_per_iteration"
+
+# Enter the gRPC repo root
+cd $(dirname $0)/../../..
+
+source tools/internal_ci/helper_scripts/prepare_build_linux_perf_rc
+
+tools/run_tests/start_port_server.py
+tools/jenkins/run_c_cpp_test.sh tools/profiling/microbenchmarks/bm_diff/bm_main.py \
+  -d origin/$ghprbTargetBranch \
+  -b bm_fullstack_trickle \
+  -l 4 \
+  -t $BENCHMARKS_TO_RUN \
+  --no-counters \
+  --pr_comment_name trickle || FAILED="true"
+
+# kill port_server.py to prevent the build from hanging
+ps aux | grep port_server\\.py | awk '{print $2}' | xargs kill -9
+
+if [ "$FAILED" != "" ]
+then
+  exit 1
+fi

tools/internal_ci/linux/pull_request/grpc_microbenchmark_diff.cfg

@@ -1,4 +1,4 @@
-# Copyright 2016 gRPC authors.
+# Copyright 2017 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,3 +11,15 @@
 # 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.
+
+# Config file for the internal CI (in protobuf text format)
+
+# Location of the continuous shell script in repository.
+build_file: "grpc/tools/internal_ci/linux/grpc_microbenchmark_diff.sh"
+timeout_mins: 120
+action {
+  define_artifacts {
+    regex: "**/*sponge_log.xml"
+    regex: "github/grpc/reports/**"
+  }
+}

tools/internal_ci/linux/pull_request/grpc_trickle_diff.cfg

@@ -1,4 +1,4 @@
-# Copyright 2016 gRPC authors.
+# Copyright 2017 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,3 +11,15 @@
 # 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.
+
+# Config file for the internal CI (in protobuf text format)
+
+# Location of the continuous shell script in repository.
+build_file: "grpc/tools/internal_ci/linux/grpc_trickle_diff.sh"
+timeout_mins: 120
+action {
+  define_artifacts {
+    regex: "**/*sponge_log.xml"
+    regex: "github/grpc/reports/**"
+  }
+}

tools/internal_ci/macos/grpc_run_tests_matrix.sh

@@ -25,6 +25,11 @@ tools/run_tests/run_tests_matrix.py $RUN_TESTS_FLAGS || FAILED="true"
 # kill port_server.py to prevent the build from hanging
 ps aux | grep port_server\\.py | awk '{print $2}' | xargs kill -9
 
+# Reveal leftover processes that might be left behind by the build
+ps aux | grep -i kbuilder
+
+echo 'Exiting gRPC main test script.'
+
 if [ "$FAILED" != "" ]
 then
   exit 1

tools/internal_ci/windows/grpc_run_tests_matrix.bat

@@ -17,8 +17,10 @@ cd /d %~dp0\..\..\..
 
 call tools/internal_ci/helper_scripts/prepare_build_windows.bat
 
-python tools/run_tests/run_tests_matrix.py %RUN_TESTS_FLAGS% || goto :error
-goto :EOF
+python tools/run_tests/run_tests_matrix.py %RUN_TESTS_FLAGS%
+set RUNTESTS_EXITCODE=%errorlevel%
 
-:error
-exit /b %errorlevel%
+@rem Reveal leftover processes that might be left behind by the build
+tasklist /V
+
+exit /b %RUNTESTS_EXITCODE%

tools/run_tests/python_utils/jobset.py

@@ -71,10 +71,8 @@ def platform_string():
 if platform_string() == 'windows':
   pass
 else:
-  have_alarm = False
   def alarm_handler(unused_signum, unused_frame):
-    global have_alarm
-    have_alarm = False
+    pass
 
   signal.signal(signal.SIGCHLD, lambda unused_signum, unused_frame: None)
   signal.signal(signal.SIGALRM, alarm_handler)
@@ -367,10 +365,9 @@ class Jobset(object):
   """Manages one run of jobs."""
 
   def __init__(self, check_cancelled, maxjobs, newline_on_success, travis,
-               stop_on_failure, add_env, quiet_success, max_time, clear_alarms):
+               stop_on_failure, add_env, quiet_success, max_time):
     self._running = set()
     self._check_cancelled = check_cancelled
-    self._clear_alarms = clear_alarms
     self._cancelled = False
     self._failures = 0
     self._completed = 0
@@ -455,10 +452,7 @@ class Jobset(object):
       if platform_string() == 'windows':
         time.sleep(0.1)
       else:
-        global have_alarm
-        if not have_alarm:
-          have_alarm = True
-          signal.alarm(10)
+        signal.alarm(10)
         signal.pause()
 
   def cancelled(self):
@@ -474,10 +468,7 @@ class Jobset(object):
     while self._running:
       if self.cancelled(): pass  # poll cancellation
       self.reap()
-    # Clear the alarms when finished to avoid a race condition causing job
-    # failures. Don't do this when running multi-VM tests because clearing
-    # the alarms causes the test to stall
-    if platform_string() != 'windows' and self._clear_alarms:
+    if platform_string() != 'windows':
       signal.alarm(0)
     return not self.cancelled() and self._failures == 0
 
@@ -507,8 +498,7 @@ def run(cmdlines,
         add_env={},
         skip_jobs=False,
         quiet_success=False,
-        max_time=-1,
-        clear_alarms=True):
+        max_time=-1):
   if skip_jobs:
     resultset = {}
     skipped_job_result = JobResult()
@@ -520,7 +510,7 @@ def run(cmdlines,
   js = Jobset(check_cancelled,
               maxjobs if maxjobs is not None else _DEFAULT_MAX_JOBS,
               newline_on_success, travis, stop_on_failure, add_env,
-              quiet_success, max_time, clear_alarms)
+              quiet_success, max_time)
   for cmdline, remaining in tag_remaining(cmdlines):
     if not js.start(cmdline):
       break

tools/run_tests/run_performance_tests.py

@@ -183,7 +183,7 @@ def archive_repo(languages):
 
   jobset.message('START', 'Archiving local repository.', do_newline=True)
   num_failures, _ = jobset.run(
-      [archive_job], newline_on_success=True, maxjobs=1, clear_alarms=False)
+      [archive_job], newline_on_success=True, maxjobs=1)
   if num_failures == 0:
     jobset.message('SUCCESS',
                    'Archive with local repository created successfully.',
@@ -215,7 +215,7 @@ def prepare_remote_hosts(hosts, prepare_local=False):
             timeout_seconds=prepare_timeout))
   jobset.message('START', 'Preparing hosts.', do_newline=True)
   num_failures, _ = jobset.run(
-      prepare_jobs, newline_on_success=True, maxjobs=10, clear_alarms=False)
+      prepare_jobs, newline_on_success=True, maxjobs=10)
   if num_failures == 0:
     jobset.message('SUCCESS',
                    'Prepare step completed successfully.',
@@ -248,7 +248,7 @@ def build_on_remote_hosts(hosts, languages=scenario_config.LANGUAGES.keys(), bui
             timeout_seconds=build_timeout))
   jobset.message('START', 'Building.', do_newline=True)
   num_failures, _ = jobset.run(
-      build_jobs, newline_on_success=True, maxjobs=10, clear_alarms=False)
+      build_jobs, newline_on_success=True, maxjobs=10)
   if num_failures == 0:
     jobset.message('SUCCESS',
                    'Built successfully.',
@@ -414,7 +414,7 @@ def run_collect_perf_profile_jobs(hosts_and_base_names, scenario_name, flame_gra
     perf_report_jobs.append(perf_report_processor_job(host, perf_base_name, output_filename, flame_graph_reports))
 
   jobset.message('START', 'Collecting perf reports from qps workers', do_newline=True)
-  failures, _ = jobset.run(perf_report_jobs, newline_on_success=True, maxjobs=1, clear_alarms=False)
+  failures, _ = jobset.run(perf_report_jobs, newline_on_success=True, maxjobs=1)
   jobset.message('END', 'Collecting perf reports from qps workers', do_newline=True)
   return failures
 
@@ -556,7 +556,7 @@ def main():
         jobs = [scenario.jobspec]
         if scenario.workers:
           jobs.append(create_quit_jobspec(scenario.workers, remote_host=args.remote_driver_host))
-        scenario_failures, resultset = jobset.run(jobs, newline_on_success=True, maxjobs=1, clear_alarms=False)
+        scenario_failures, resultset = jobset.run(jobs, newline_on_success=True, maxjobs=1)
         total_scenario_failures += scenario_failures
         merged_resultset = dict(itertools.chain(six.iteritems(merged_resultset),
                                                 six.iteritems(resultset)))