Merge pull request #16492 from vjpai/client_callback

EXPERIMENTAL: C++  generic client-side unary callback API

7 files changed, 253 insertions, 5 deletions

diff --git a/include/grpcpp/impl/codegen/call.h b/include/grpcpp/impl/codegen/call.h
index e94adada94..7cadea0055 100644
--- a/include/grpcpp/impl/codegen/call.h
+++ b/include/grpcpp/impl/codegen/call.h
@@ -599,6 +599,11 @@ class CallOpSetInterface : public CompletionQueueTag {
   /// Fills in grpc_op, starting from ops[*nops] and moving
   /// upwards.
   virtual void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) = 0;
+
+  /// Get the tag to be used at the core completion queue. Generally, the
+  /// value of cq_tag will be "this". However, it can be overridden if we
+  /// want core to process the tag differently (e.g., as a core callback)
+  virtual void* cq_tag() = 0;
 };
 
 /// Primary implementation of CallOpSetInterface.
@@ -618,7 +623,7 @@ class CallOpSet : public CallOpSetInterface,
                   public Op5,
                   public Op6 {
  public:
-  CallOpSet() : return_tag_(this), call_(nullptr) {}
+  CallOpSet() : cq_tag_(this), return_tag_(this), call_(nullptr) {}
   void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) override {
     this->Op1::AddOp(ops, nops);
     this->Op2::AddOp(ops, nops);
@@ -645,7 +650,16 @@ class CallOpSet : public CallOpSetInterface,
 
   void set_output_tag(void* return_tag) { return_tag_ = return_tag; }
 
+  void* cq_tag() override { return cq_tag_; }
+
+  /// set_cq_tag is used to provide a different core CQ tag than "this".
+  /// This is used for callback-based tags, where the core tag is the core
+  /// callback function. It does not change the use or behavior of any other
+  /// function (such as FinalizeResult)
+  void set_cq_tag(void* cq_tag) { cq_tag_ = cq_tag; }
+
  private:
+  void* cq_tag_;
   void* return_tag_;
   grpc_call* call_;
 };
diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h
new file mode 100644
index 0000000000..68c318d2b4
--- /dev/null
+++ b/include/grpcpp/impl/codegen/callback_common.h
@@ -0,0 +1,103 @@
+/*
+ *
+ * Copyright 2018 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.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
+#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
+
+#include <functional>
+
+#include <grpcpp/impl/codegen/call.h>
+#include <grpcpp/impl/codegen/channel_interface.h>
+#include <grpcpp/impl/codegen/config.h>
+#include <grpcpp/impl/codegen/core_codegen_interface.h>
+#include <grpcpp/impl/codegen/status.h>
+
+// Forward declarations
+namespace grpc_core {
+class CQCallbackInterface;
+};
+
+namespace grpc {
+namespace internal {
+
+class CallbackWithStatusTag {
+ public:
+  // always allocated against a call arena, no memory free required
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(CallbackWithStatusTag));
+  }
+
+  // This operator should never be called as the memory should be freed as part
+  // of the arena destruction. It only exists to provide a matching operator
+  // delete to the operator new so that some compilers will not complain (see
+  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
+  // there are no tests catching the compiler warning.
+  static void operator delete(void*, void*) { assert(0); }
+
+  CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f,
+                        CompletionQueueTag* ops);
+  ~CallbackWithStatusTag() {}
+  void* tag() { return static_cast<void*>(impl_); }
+  Status* status_ptr() { return status_; }
+  CompletionQueueTag* ops() { return ops_; }
+
+  // force_run can not be performed on a tag if operations using this tag
+  // have been sent to PerformOpsOnCall. It is intended for error conditions
+  // that are detected before the operations are internally processed.
+  void force_run(Status s);
+
+ private:
+  grpc_core::CQCallbackInterface* impl_;
+  Status* status_;
+  CompletionQueueTag* ops_;
+};
+
+class CallbackWithSuccessTag {
+ public:
+  // always allocated against a call arena, no memory free required
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(CallbackWithSuccessTag));
+  }
+
+  // This operator should never be called as the memory should be freed as part
+  // of the arena destruction. It only exists to provide a matching operator
+  // delete to the operator new so that some compilers will not complain (see
+  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
+  // there are no tests catching the compiler warning.
+  static void operator delete(void*, void*) { assert(0); }
+
+  CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f,
+                         CompletionQueueTag* ops);
+
+  void* tag() { return static_cast<void*>(impl_); }
+  CompletionQueueTag* ops() { return ops_; }
+
+  // force_run can not be performed on a tag if operations using this tag
+  // have been sent to PerformOpsOnCall. It is intended for error conditions
+  // that are detected before the operations are internally processed.
+  void force_run(bool ok);
+
+ private:
+  grpc_core::CQCallbackInterface* impl_;
+  CompletionQueueTag* ops_;
+};
+
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h
index ec1c6c25d8..b257acc1ab 100644
--- a/include/grpcpp/impl/codegen/channel_interface.h
+++ b/include/grpcpp/impl/codegen/channel_interface.h
@@ -41,6 +41,8 @@ class CallOpSetInterface;
 class RpcMethod;
 template <class InputMessage, class OutputMessage>
 class BlockingUnaryCallImpl;
+template <class InputMessage, class OutputMessage>
+class CallbackUnaryCallImpl;
 template <class R>
 class ClientAsyncReaderFactory;
 template <class W>
@@ -103,6 +105,8 @@ class ChannelInterface {
   friend class ::grpc::internal::ClientAsyncResponseReaderFactory;
   template <class InputMessage, class OutputMessage>
   friend class ::grpc::internal::BlockingUnaryCallImpl;
+  template <class InputMessage, class OutputMessage>
+  friend class ::grpc::internal::CallbackUnaryCallImpl;
   friend class ::grpc::internal::RpcMethod;
   virtual internal::Call CreateCall(const internal::RpcMethod& method,
                                     ClientContext* context,
@@ -115,6 +119,16 @@ class ChannelInterface {
                                        CompletionQueue* cq, void* tag) = 0;
   virtual bool WaitForStateChangeImpl(grpc_connectivity_state last_observed,
                                       gpr_timespec deadline) = 0;
+
+  // EXPERIMENTAL
+  // A method to get the callbackable completion queue associated with this
+  // channel. If the return value is nullptr, this channel doesn't support
+  // callback operations.
+  // TODO(vjpai): Consider a better default like using a global CQ
+  // Returns nullptr (rather than being pure) since this is a new method
+  // and adding a new pure method to an interface would be a breaking change
+  // (even though this is private and non-API)
+  virtual CompletionQueue* CallbackCQ() { return nullptr; }
 };
 }  // namespace grpc
 
diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h
new file mode 100644
index 0000000000..fc81c8aa0a
--- /dev/null
+++ b/include/grpcpp/impl/codegen/client_callback.h
@@ -0,0 +1,95 @@
+/*
+ *
+ * Copyright 2018 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.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
+#define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
+
+#include <functional>
+
+#include <grpcpp/impl/codegen/call.h>
+#include <grpcpp/impl/codegen/callback_common.h>
+#include <grpcpp/impl/codegen/channel_interface.h>
+#include <grpcpp/impl/codegen/config.h>
+#include <grpcpp/impl/codegen/core_codegen_interface.h>
+#include <grpcpp/impl/codegen/status.h>
+
+namespace grpc {
+
+class Channel;
+class ClientContext;
+class CompletionQueue;
+
+namespace internal {
+class RpcMethod;
+
+/// Perform a callback-based unary call
+/// TODO(vjpai): Combine as much as possible with the blocking unary call code
+template <class InputMessage, class OutputMessage>
+void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method,
+                       ClientContext* context, const InputMessage* request,
+                       OutputMessage* result,
+                       std::function<void(Status)> on_completion) {
+  CallbackUnaryCallImpl<InputMessage, OutputMessage> x(
+      channel, method, context, request, result, on_completion);
+}
+
+template <class InputMessage, class OutputMessage>
+class CallbackUnaryCallImpl {
+ public:
+  CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method,
+                        ClientContext* context, const InputMessage* request,
+                        OutputMessage* result,
+                        std::function<void(Status)> on_completion) {
+    CompletionQueue* cq = channel->CallbackCQ();
+    GPR_CODEGEN_ASSERT(cq != nullptr);
+    Call call(channel->CreateCall(method, context, cq));
+
+    using FullCallOpSet =
+        CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+                  CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
+                  CallOpClientSendClose, CallOpClientRecvStatus>;
+
+    auto* ops = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call.call(), sizeof(FullCallOpSet))) FullCallOpSet;
+
+    auto* tag = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call.call(), sizeof(CallbackWithStatusTag)))
+        CallbackWithStatusTag(call.call(), on_completion, ops);
+
+    // TODO(vjpai): Unify code with sync API as much as possible
+    Status s = ops->SendMessage(*request);
+    if (!s.ok()) {
+      tag->force_run(s);
+      return;
+    }
+    ops->SendInitialMetadata(context->send_initial_metadata_,
+                             context->initial_metadata_flags());
+    ops->RecvInitialMetadata(context);
+    ops->RecvMessage(result);
+    ops->AllowNoMessage();
+    ops->ClientSendClose();
+    ops->ClientRecvStatus(context, tag->status_ptr());
+    ops->set_cq_tag(tag->tag());
+    call.PerformOps(ops);
+  }
+};
+
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
diff --git a/include/grpcpp/impl/codegen/client_context.h b/include/grpcpp/impl/codegen/client_context.h
index c6c9540950..46635a541a 100644
--- a/include/grpcpp/impl/codegen/client_context.h
+++ b/include/grpcpp/impl/codegen/client_context.h
@@ -68,6 +68,8 @@ class CallOpClientRecvStatus;
 class CallOpRecvInitialMetadata;
 template <class InputMessage, class OutputMessage>
 class BlockingUnaryCallImpl;
+template <class InputMessage, class OutputMessage>
+class CallbackUnaryCallImpl;
 }  // namespace internal
 
 template <class R>
@@ -389,6 +391,8 @@ class ClientContext {
   friend class ::grpc::ClientAsyncResponseReader;
   template <class InputMessage, class OutputMessage>
   friend class ::grpc::internal::BlockingUnaryCallImpl;
+  template <class InputMessage, class OutputMessage>
+  friend class ::grpc::internal::CallbackUnaryCallImpl;
 
   // Used by friend class CallOpClientRecvStatus
   void set_debug_error_string(const grpc::string& debug_error_string) {
diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h
index 3f7d4fb765..f52f9a53be 100644
--- a/include/grpcpp/impl/codegen/completion_queue.h
+++ b/include/grpcpp/impl/codegen/completion_queue.h
@@ -274,6 +274,9 @@ class CompletionQueue : private GrpcLibraryCodegen {
   template <class InputMessage, class OutputMessage>
   friend class ::grpc::internal::BlockingUnaryCallImpl;
 
+  // Friends that need access to constructor for callback CQ
+  friend class ::grpc::Channel;
+
   /// EXPERIMENTAL
   /// Creates a Thread Local cache to store the first event
   /// On this completion queue queued from this thread.  Once
diff --git a/include/grpcpp/impl/codegen/completion_queue_tag.h b/include/grpcpp/impl/codegen/completion_queue_tag.h
index ffb642c56b..304386a9ec 100644
--- a/include/grpcpp/impl/codegen/completion_queue_tag.h
+++ b/include/grpcpp/impl/codegen/completion_queue_tag.h
@@ -26,10 +26,25 @@ namespace internal {
 class CompletionQueueTag {
  public:
   virtual ~CompletionQueueTag() {}
-  /// Called prior to returning from Next(), return value is the status of the
-  /// operation (return status is the default thing to do). If this function
-  /// returns false, the tag is dropped and not returned from the completion
-  /// queue
+
+  /// FinalizeResult must be called before informing user code that the
+  /// operation bound to the underlying core completion queue tag has
+  /// completed. In practice, this means:
+  ///
+  ///   1. For the sync API - before returning from Pluck
+  ///   2. For the CQ-based async API - before returning from Next
+  ///   3. For the callback-based API - before invoking the user callback
+  ///
+  /// This is the method that translates from core-side tag/status to
+  /// C++ API-observable tag/status.
+  ///
+  /// The return value is the status of the operation (returning status is the
+  /// general behavior of this function). If this function returns false, the
+  /// tag is dropped and not returned from the completion queue: this concept is
+  /// for events that are observed at core but not requested by the user
+  /// application (e.g., server shutdown, for server unimplemented method
+  /// responses, or for cases where a server-side RPC doesn't have a completion
+  /// notification registered using AsyncNotifyWhenDone)
   virtual bool FinalizeResult(void** tag, bool* status) = 0;
 };
 }  // namespace internal