Merge branch 'master' into statuscaution

7 files changed, 239 insertions, 4 deletions

diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc
index 39b891c2e1..ad71286e05 100644
--- a/src/cpp/client/channel_cc.cc
+++ b/src/cpp/client/channel_cc.cc
@@ -42,8 +42,10 @@
 #include <grpcpp/support/time.h>
 #include "src/core/lib/gpr/env.h"
 #include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gprpp/memory.h"
 #include "src/core/lib/gprpp/thd.h"
 #include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/surface/completion_queue.h"
 
 namespace grpc {
 
@@ -53,7 +55,12 @@ Channel::Channel(const grpc::string& host, grpc_channel* channel)
   g_gli_initializer.summon();
 }
 
-Channel::~Channel() { grpc_channel_destroy(c_channel_); }
+Channel::~Channel() {
+  grpc_channel_destroy(c_channel_);
+  if (callback_cq_ != nullptr) {
+    callback_cq_->Shutdown();
+  }
+}
 
 namespace {
 
@@ -135,8 +142,8 @@ void Channel::PerformOpsOnCall(internal::CallOpSetInterface* ops,
   size_t nops = 0;
   grpc_op cops[MAX_OPS];
   ops->FillOps(call->call(), cops, &nops);
-  GPR_ASSERT(GRPC_CALL_OK ==
-             grpc_call_start_batch(call->call(), cops, nops, ops, nullptr));
+  GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call->call(), cops, nops,
+                                                   ops->cq_tag(), nullptr));
 }
 
 void* Channel::RegisterMethod(const char* method) {
@@ -185,4 +192,39 @@ bool Channel::WaitForStateChangeImpl(grpc_connectivity_state last_observed,
   return ok;
 }
 
+namespace {
+class ShutdownCallback : public grpc_core::CQCallbackInterface {
+ public:
+  // TakeCQ takes ownership of the cq into the shutdown callback
+  // so that the shutdown callback will be responsible for destroying it
+  void TakeCQ(CompletionQueue* cq) { cq_ = cq; }
+
+  // The Run function will get invoked by the completion queue library
+  // when the shutdown is actually complete
+  void Run(bool) override {
+    delete cq_;
+    grpc_core::Delete(this);
+  }
+
+ private:
+  CompletionQueue* cq_ = nullptr;
+};
+}  // namespace
+
+CompletionQueue* Channel::CallbackCQ() {
+  // TODO(vjpai): Consider using a single global CQ for the default CQ
+  // if there is no explicit per-channel CQ registered
+  std::lock_guard<std::mutex> l(mu_);
+  if (callback_cq_ == nullptr) {
+    auto* shutdown_callback = grpc_core::New<ShutdownCallback>();
+    callback_cq_ = new CompletionQueue(grpc_completion_queue_attributes{
+        GRPC_CQ_CURRENT_VERSION, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING,
+        shutdown_callback});
+
+    // Transfer ownership of the new cq to its own shutdown callback
+    shutdown_callback->TakeCQ(callback_cq_);
+  }
+  return callback_cq_;
+}
+
 }  // namespace grpc
diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc
index 67ef46bebe..87902b26f0 100644
--- a/src/cpp/client/generic_stub.cc
+++ b/src/cpp/client/generic_stub.cc
@@ -16,9 +16,11 @@
  *
  */
 
-#include <grpcpp/generic/generic_stub.h>
+#include <functional>
 
+#include <grpcpp/generic/generic_stub.h>
 #include <grpcpp/impl/rpc_method.h>
+#include <grpcpp/support/client_callback.h>
 
 namespace grpc {
 
@@ -60,4 +62,14 @@ std::unique_ptr<GenericClientAsyncResponseReader> GenericStub::PrepareUnaryCall(
           context, request, false));
 }
 
+void GenericStub::experimental_type::UnaryCall(
+    ClientContext* context, const grpc::string& method,
+    const ByteBuffer* request, ByteBuffer* response,
+    std::function<void(Status)> on_completion) {
+  internal::CallbackUnaryCall(
+      stub_->channel_.get(),
+      internal::RpcMethod(method.c_str(), internal::RpcMethod::NORMAL_RPC),
+      context, request, response, std::move(on_completion));
+}
+
 }  // namespace grpc
diff --git a/src/cpp/common/callback_common.cc b/src/cpp/common/callback_common.cc
new file mode 100644
index 0000000000..ae47901f1b
--- /dev/null
+++ b/src/cpp/common/callback_common.cc
@@ -0,0 +1,131 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include <functional>
+
+#include <grpcpp/impl/codegen/callback_common.h>
+#include <grpcpp/impl/codegen/status.h>
+
+#include "src/core/lib/gprpp/memory.h"
+#include "src/core/lib/surface/completion_queue.h"
+
+namespace grpc {
+namespace internal {
+
+namespace {
+class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface {
+ public:
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(CallbackWithSuccessImpl));
+  }
+
+  // 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); }
+
+  CallbackWithSuccessImpl(grpc_call* call, CallbackWithSuccessTag* parent,
+                          std::function<void(bool)> f)
+      : call_(call), parent_(parent), func_(std::move(f)) {
+    grpc_call_ref(call);
+  }
+
+  void Run(bool ok) override {
+    void* ignored = parent_->ops();
+    bool new_ok = ok;
+    GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok));
+    GPR_ASSERT(ignored == parent_->ops());
+    func_(ok);
+    func_ = nullptr;  // release the function
+    grpc_call_unref(call_);
+  }
+
+ private:
+  grpc_call* call_;
+  CallbackWithSuccessTag* parent_;
+  std::function<void(bool)> func_;
+};
+
+class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface {
+ public:
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(CallbackWithStatusImpl));
+  }
+
+  // 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); }
+
+  CallbackWithStatusImpl(grpc_call* call, CallbackWithStatusTag* parent,
+                         std::function<void(Status)> f)
+      : call_(call), parent_(parent), func_(std::move(f)), status_() {
+    grpc_call_ref(call);
+  }
+
+  void Run(bool ok) override {
+    void* ignored = parent_->ops();
+
+    GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok));
+    GPR_ASSERT(ignored == parent_->ops());
+
+    func_(status_);
+    func_ = nullptr;  // release the function
+    grpc_call_unref(call_);
+  }
+  Status* status_ptr() { return &status_; }
+
+ private:
+  grpc_call* call_;
+  CallbackWithStatusTag* parent_;
+  std::function<void(Status)> func_;
+  Status status_;
+};
+
+}  // namespace
+
+CallbackWithSuccessTag::CallbackWithSuccessTag(grpc_call* call,
+                                               std::function<void(bool)> f,
+                                               CompletionQueueTag* ops)
+    : impl_(new (grpc_call_arena_alloc(call, sizeof(CallbackWithSuccessImpl)))
+                CallbackWithSuccessImpl(call, this, std::move(f))),
+      ops_(ops) {}
+
+void CallbackWithSuccessTag::force_run(bool ok) { impl_->Run(ok); }
+
+CallbackWithStatusTag::CallbackWithStatusTag(grpc_call* call,
+                                             std::function<void(Status)> f,
+                                             CompletionQueueTag* ops)
+    : ops_(ops) {
+  auto* impl = new (grpc_call_arena_alloc(call, sizeof(CallbackWithStatusImpl)))
+      CallbackWithStatusImpl(call, this, std::move(f));
+  impl_ = impl;
+  status_ = impl->status_ptr();
+}
+
+void CallbackWithStatusTag::force_run(Status s) {
+  *status_ = std::move(s);
+  impl_->Run(true);
+}
+
+}  // namespace internal
+}  // namespace grpc
diff --git a/src/cpp/server/channelz/channelz_service.cc b/src/cpp/server/channelz/channelz_service.cc
index 77c175e5b8..e096c1f421 100644
--- a/src/cpp/server/channelz/channelz_service.cc
+++ b/src/cpp/server/channelz/channelz_service.cc
@@ -32,6 +32,25 @@ Status ChannelzService::GetTopChannels(
     ServerContext* unused, const channelz::v1::GetTopChannelsRequest* request,
     channelz::v1::GetTopChannelsResponse* response) {
   char* json_str = grpc_channelz_get_top_channels(request->start_channel_id());
+  if (json_str == nullptr) {
+    return Status(INTERNAL, "grpc_channelz_get_top_channels returned null");
+  }
+  google::protobuf::util::Status s =
+      google::protobuf::util::JsonStringToMessage(json_str, response);
+  gpr_free(json_str);
+  if (s != google::protobuf::util::Status::OK) {
+    return Status(INTERNAL, s.ToString());
+  }
+  return Status::OK;
+}
+
+Status ChannelzService::GetServers(
+    ServerContext* unused, const channelz::v1::GetServersRequest* request,
+    channelz::v1::GetServersResponse* response) {
+  char* json_str = grpc_channelz_get_servers(request->start_server_id());
+  if (json_str == nullptr) {
+    return Status(INTERNAL, "grpc_channelz_get_servers returned null");
+  }
   google::protobuf::util::Status s =
       google::protobuf::util::JsonStringToMessage(json_str, response);
   gpr_free(json_str);
@@ -45,6 +64,25 @@ Status ChannelzService::GetChannel(
     ServerContext* unused, const channelz::v1::GetChannelRequest* request,
     channelz::v1::GetChannelResponse* response) {
   char* json_str = grpc_channelz_get_channel(request->channel_id());
+  if (json_str == nullptr) {
+    return Status(NOT_FOUND, "No object found for that ChannelId");
+  }
+  google::protobuf::util::Status s =
+      google::protobuf::util::JsonStringToMessage(json_str, response);
+  gpr_free(json_str);
+  if (s != google::protobuf::util::Status::OK) {
+    return Status(INTERNAL, s.ToString());
+  }
+  return Status::OK;
+}
+
+Status ChannelzService::GetSubchannel(
+    ServerContext* unused, const channelz::v1::GetSubchannelRequest* request,
+    channelz::v1::GetSubchannelResponse* response) {
+  char* json_str = grpc_channelz_get_subchannel(request->subchannel_id());
+  if (json_str == nullptr) {
+    return Status(NOT_FOUND, "No object found for that SubchannelId");
+  }
   google::protobuf::util::Status s =
       google::protobuf::util::JsonStringToMessage(json_str, response);
   gpr_free(json_str);
diff --git a/src/cpp/server/channelz/channelz_service.h b/src/cpp/server/channelz/channelz_service.h
index f619ea49e0..9e0b5b6ead 100644
--- a/src/cpp/server/channelz/channelz_service.h
+++ b/src/cpp/server/channelz/channelz_service.h
@@ -32,10 +32,18 @@ class ChannelzService final : public channelz::v1::Channelz::Service {
   Status GetTopChannels(
       ServerContext* unused, const channelz::v1::GetTopChannelsRequest* request,
       channelz::v1::GetTopChannelsResponse* response) override;
+  // implementation of GetServers rpc
+  Status GetServers(ServerContext* unused,
+                    const channelz::v1::GetServersRequest* request,
+                    channelz::v1::GetServersResponse* response) override;
   // implementation of GetChannel rpc
   Status GetChannel(ServerContext* unused,
                     const channelz::v1::GetChannelRequest* request,
                     channelz::v1::GetChannelResponse* response) override;
+  // implementation of GetSubchannel rpc
+  Status GetSubchannel(ServerContext* unused,
+                       const channelz::v1::GetSubchannelRequest* request,
+                       channelz::v1::GetSubchannelResponse* response) override;
 };
 
 }  // namespace grpc
diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc
index b8ba7042d9..36f7eb81f9 100644
--- a/src/cpp/server/server_cc.cc
+++ b/src/cpp/server/server_cc.cc
@@ -657,6 +657,7 @@ void Server::PerformOpsOnCall(internal::CallOpSetInterface* ops,
   size_t nops = 0;
   grpc_op cops[MAX_OPS];
   ops->FillOps(call->call(), cops, &nops);
+  // TODO(vjpai): Use ops->cq_tag once this case supports callbacks
   auto result = grpc_call_start_batch(call->call(), cops, nops, ops, nullptr);
   if (result != GRPC_CALL_OK) {
     gpr_log(GPR_ERROR, "Fatal: grpc_call_start_batch returned %d", result);
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index bf0c027cda..b7254b6bb9 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -61,6 +61,9 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface {
     tag_ = tag;
   }
 
+  /// TODO(vjpai): Allow override of cq_tag if appropriate for callback API
+  void* cq_tag() override { return this; }
+
   void Unref();
 
  private: