Merge pull request #18165 from vjpai/raw_req_client

Support use of ByteBuffer for request-side of code-gen unary

include/grpcpp/impl/codegen/byte_buffer.h

@@ -96,7 +96,7 @@ class ByteBuffer final {
   /// \a buf. Wrapper of core function grpc_byte_buffer_copy . This is not
   /// a deep copy; it is just a referencing. As a result, its performance is
   /// size-independent.
-  ByteBuffer(const ByteBuffer& buf);
+  ByteBuffer(const ByteBuffer& buf) : buffer_(nullptr) { operator=(buf); }
 
   ~ByteBuffer() {
     if (buffer_) {
@@ -107,7 +107,16 @@ class ByteBuffer final {
   /// Wrapper of core function grpc_byte_buffer_copy . This is not
   /// a deep copy; it is just a referencing. As a result, its performance is
   /// size-independent.
-  ByteBuffer& operator=(const ByteBuffer&);
+  ByteBuffer& operator=(const ByteBuffer& buf) {
+    if (this != &buf) {
+      Clear();  // first remove existing data
+    }
+    if (buf.buffer_) {
+      // then copy
+      buffer_ = g_core_codegen_interface->grpc_byte_buffer_copy(buf.buffer_);
+    }
+    return *this;
+  }
 
   /// Dump (read) the buffer contents into \a slices.
   Status Dump(std::vector<Slice>* slices) const;
@@ -215,7 +224,7 @@ class SerializationTraits<ByteBuffer, void> {
                           bool* own_buffer) {
     *buffer = source;
     *own_buffer = true;
-    return Status::OK;
+    return g_core_codegen_interface->ok();
   }
 };
 

src/compiler/cpp_generator.cc

@@ -580,6 +580,10 @@ void PrintHeaderClientMethodCallbackInterfaces(
                    "virtual void $Method$(::grpc::ClientContext* context, "
                    "const $Request$* request, $Response$* response, "
                    "std::function<void(::grpc::Status)>) = 0;\n");
+    printer->Print(*vars,
+                   "virtual void $Method$(::grpc::ClientContext* context, "
+                   "const ::grpc::ByteBuffer* request, $Response$* response, "
+                   "std::function<void(::grpc::Status)>) = 0;\n");
   } else if (ClientOnlyStreaming(method)) {
     printer->Print(*vars,
                    "virtual void $Method$(::grpc::ClientContext* context, "
@@ -642,6 +646,10 @@ void PrintHeaderClientMethodCallback(grpc_generator::Printer* printer,
                    "void $Method$(::grpc::ClientContext* context, "
                    "const $Request$* request, $Response$* response, "
                    "std::function<void(::grpc::Status)>) override;\n");
+    printer->Print(*vars,
+                   "void $Method$(::grpc::ClientContext* context, "
+                   "const ::grpc::ByteBuffer* request, $Response$* response, "
+                   "std::function<void(::grpc::Status)>) override;\n");
   } else if (ClientOnlyStreaming(method)) {
     printer->Print(*vars,
                    "void $Method$(::grpc::ClientContext* context, "
@@ -1643,6 +1651,16 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
                    "(stub_->channel_.get(), stub_->rpcmethod_$Method$_, "
                    "context, request, response, std::move(f));\n}\n\n");
 
+    printer->Print(*vars,
+                   "void $ns$$Service$::Stub::experimental_async::$Method$("
+                   "::grpc::ClientContext* context, "
+                   "const ::grpc::ByteBuffer* request, $Response$* response, "
+                   "std::function<void(::grpc::Status)> f) {\n");
+    printer->Print(*vars,
+                   "  return ::grpc::internal::CallbackUnaryCall"
+                   "(stub_->channel_.get(), stub_->rpcmethod_$Method$_, "
+                   "context, request, response, std::move(f));\n}\n\n");
+
     for (auto async_prefix : async_prefixes) {
       (*vars)["AsyncPrefix"] = async_prefix.prefix;
       (*vars)["AsyncStart"] = async_prefix.start;

src/cpp/util/byte_buffer_cc.cc

@@ -43,18 +43,4 @@ Status ByteBuffer::Dump(std::vector<Slice>* slices) const {
   return Status::OK;
 }
 
-ByteBuffer::ByteBuffer(const ByteBuffer& buf) : buffer_(nullptr) {
-  operator=(buf);
-}
-
-ByteBuffer& ByteBuffer::operator=(const ByteBuffer& buf) {
-  if (this != &buf) {
-    Clear();  // first remove existing data
-  }
-  if (buf.buffer_) {
-    buffer_ = grpc_byte_buffer_copy(buf.buffer_);  // then copy
-  }
-  return *this;
-}
-
 }  // namespace grpc

test/cpp/codegen/compiler_test_golden

@@ -113,6 +113,7 @@ class ServiceA final {
       virtual ~experimental_async_interface() {}
       // MethodA1 leading comment 1
       virtual void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) = 0;
+      virtual void MethodA1(::grpc::ClientContext* context, const ::grpc::ByteBuffer* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) = 0;
       // MethodA1 trailing comment 1
       // MethodA2 detached leading comment 1
       //
@@ -182,6 +183,7 @@ class ServiceA final {
       public StubInterface::experimental_async_interface {
      public:
       void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override;
+      void MethodA1(::grpc::ClientContext* context, const ::grpc::ByteBuffer* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override;
       void MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor< ::grpc::testing::Request>* reactor) override;
       void MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor< ::grpc::testing::Response>* reactor) override;
       void MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::grpc::testing::Request,::grpc::testing::Response>* reactor) override;
@@ -714,6 +716,7 @@ class ServiceB final {
       virtual ~experimental_async_interface() {}
       // MethodB1 leading comment 1
       virtual void MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) = 0;
+      virtual void MethodB1(::grpc::ClientContext* context, const ::grpc::ByteBuffer* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) = 0;
       // MethodB1 trailing comment 1
     };
     virtual class experimental_async_interface* experimental_async() { return nullptr; }
@@ -735,6 +738,7 @@ class ServiceB final {
       public StubInterface::experimental_async_interface {
      public:
       void MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override;
+      void MethodB1(::grpc::ClientContext* context, const ::grpc::ByteBuffer* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override;
      private:
       friend class Stub;
       explicit experimental_async(Stub* stub): stub_(stub) { }

test/cpp/end2end/client_callback_end2end_test.cc

@@ -220,6 +220,36 @@ class ClientCallbackEnd2endTest
     }
   }
 
+  void SendRpcsRawReq(int num_rpcs) {
+    grpc::string test_string("Hello raw world.");
+    EchoRequest request;
+    request.set_message(test_string);
+    std::unique_ptr<ByteBuffer> send_buf = SerializeToByteBuffer(&request);
+
+    for (int i = 0; i < num_rpcs; i++) {
+      EchoResponse response;
+      ClientContext cli_ctx;
+
+      std::mutex mu;
+      std::condition_variable cv;
+      bool done = false;
+      stub_->experimental_async()->Echo(
+          &cli_ctx, send_buf.get(), &response,
+          [&request, &response, &done, &mu, &cv](Status s) {
+            GPR_ASSERT(s.ok());
+
+            EXPECT_EQ(request.message(), response.message());
+            std::lock_guard<std::mutex> l(mu);
+            done = true;
+            cv.notify_one();
+          });
+      std::unique_lock<std::mutex> l(mu);
+      while (!done) {
+        cv.wait(l);
+      }
+    }
+  }
+
   void SendRpcsGeneric(int num_rpcs, bool maybe_except) {
     const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
     grpc::string test_string("");
@@ -347,6 +377,12 @@ TEST_P(ClientCallbackEnd2endTest, SequentialRpcs) {
   SendRpcs(10, false);
 }
 
+TEST_P(ClientCallbackEnd2endTest, SequentialRpcsRawReq) {
+  MAYBE_SKIP_TEST;
+  ResetStub();
+  SendRpcsRawReq(10);
+}
+
 TEST_P(ClientCallbackEnd2endTest, SendClientInitialMetadata) {
   MAYBE_SKIP_TEST;
   ResetStub();

test/cpp/microbenchmarks/helpers.cc

@@ -20,6 +20,17 @@
 
 #include "test/cpp/microbenchmarks/helpers.h"
 
+static grpc::internal::GrpcLibraryInitializer g_gli_initializer;
+
+Library::Library() {
+  g_gli_initializer.summon();
+#ifdef GPR_LOW_LEVEL_COUNTERS
+  grpc_memory_counters_init();
+#endif
+  init_lib_.init();
+  rq_ = grpc_resource_quota_create("bm");
+}
+
 void TrackCounters::Finish(benchmark::State& state) {
   std::ostringstream out;
   for (const auto& l : labels_) {

test/cpp/microbenchmarks/helpers.h

@@ -39,13 +39,7 @@ class Library {
   grpc_resource_quota* rq() { return rq_; }
 
  private:
-  Library() {
-#ifdef GPR_LOW_LEVEL_COUNTERS
-    grpc_memory_counters_init();
-#endif
-    init_lib_.init();
-    rq_ = grpc_resource_quota_create("bm");
-  }
+  Library();
 
   ~Library() { init_lib_.shutdown(); }