1 files changed, 25 insertions, 179 deletions

diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc
index 19a25c838f..9df531066e 100644
--- a/src/cpp/client/channel_cc.cc
+++ b/src/cpp/client/channel_cc.cc
@@ -48,187 +48,13 @@
 
 namespace grpc {
 
-namespace {
-int kConnectivityCheckIntervalMsec = 500;
-void WatchStateChange(void* arg);
-
-class TagSaver final : public CompletionQueueTag {
- public:
-  explicit TagSaver(void* tag) : tag_(tag) {}
-  ~TagSaver() override {}
-  bool FinalizeResult(void** tag, bool* status) override {
-    *tag = tag_;
-    delete this;
-    return true;
-  }
-
- private:
-  void* tag_;
-};
-
-// Constantly watches channel connectivity status to reconnect a transiently
-// disconnected channel. This is a temporary work-around before we have retry
-// support.
-class ChannelConnectivityWatcher : private GrpcLibraryCodegen {
- public:
-  static void StartWatching(grpc_channel* channel) {
-    if (!IsDisabled()) {
-      std::unique_lock<std::mutex> lock(g_watcher_mu_);
-      if (g_watcher_ == nullptr) {
-        g_watcher_ = new ChannelConnectivityWatcher();
-      }
-      g_watcher_->StartWatchingLocked(channel);
-    }
-  }
-
-  static void StopWatching() {
-    if (!IsDisabled()) {
-      std::unique_lock<std::mutex> lock(g_watcher_mu_);
-      if (g_watcher_->StopWatchingLocked()) {
-        delete g_watcher_;
-        g_watcher_ = nullptr;
-      }
-    }
-  }
-
- private:
-  ChannelConnectivityWatcher() : channel_count_(0), shutdown_(false) {
-    gpr_ref_init(&ref_, 0);
-    gpr_thd_options options = gpr_thd_options_default();
-    gpr_thd_options_set_joinable(&options);
-    gpr_thd_new(&thd_id_, &WatchStateChange, this, &options);
-  }
-
-  static bool IsDisabled() {
-    char* env = gpr_getenv("GRPC_DISABLE_CHANNEL_CONNECTIVITY_WATCHER");
-    bool disabled = gpr_is_true(env);
-    gpr_free(env);
-    return disabled;
-  }
-
-  void WatchStateChangeImpl() {
-    bool ok = false;
-    void* tag = NULL;
-    CompletionQueue::NextStatus status = CompletionQueue::GOT_EVENT;
-    while (true) {
-      {
-        std::unique_lock<std::mutex> lock(shutdown_mu_);
-        if (shutdown_) {
-          // Drain cq_ if the watcher is shutting down
-          status = cq_.AsyncNext(&tag, &ok, gpr_inf_future(GPR_CLOCK_REALTIME));
-        } else {
-          status = cq_.AsyncNext(&tag, &ok, gpr_inf_past(GPR_CLOCK_REALTIME));
-          // Make sure we've seen 2 TIMEOUTs before going to sleep
-          if (status == CompletionQueue::TIMEOUT) {
-            status = cq_.AsyncNext(&tag, &ok, gpr_inf_past(GPR_CLOCK_REALTIME));
-            if (status == CompletionQueue::TIMEOUT) {
-              shutdown_cv_.wait_for(lock, std::chrono::milliseconds(
-                                              kConnectivityCheckIntervalMsec));
-              continue;
-            }
-          }
-        }
-      }
-      ChannelState* channel_state = static_cast<ChannelState*>(tag);
-      channel_state->state =
-          grpc_channel_check_connectivity_state(channel_state->channel, false);
-      if (channel_state->state == GRPC_CHANNEL_SHUTDOWN) {
-        void* shutdown_tag = NULL;
-        channel_state->shutdown_cq.Next(&shutdown_tag, &ok);
-        delete channel_state;
-        if (gpr_unref(&ref_)) {
-          break;
-        }
-      } else {
-        TagSaver* tag_saver = new TagSaver(channel_state);
-        grpc_channel_watch_connectivity_state(
-            channel_state->channel, channel_state->state,
-            gpr_inf_future(GPR_CLOCK_REALTIME), cq_.cq(), tag_saver);
-      }
-    }
-  }
-
-  void StartWatchingLocked(grpc_channel* channel) {
-    if (thd_id_ != 0) {
-      gpr_ref(&ref_);
-      ++channel_count_;
-      ChannelState* channel_state = new ChannelState(channel);
-      // The first grpc_channel_watch_connectivity_state() is not used to
-      // monitor the channel state change, but to hold a reference of the
-      // c channel. So that WatchStateChangeImpl() can observe state ==
-      // GRPC_CHANNEL_SHUTDOWN before the channel gets destroyed.
-      grpc_channel_watch_connectivity_state(
-          channel_state->channel, channel_state->state,
-          gpr_inf_future(GPR_CLOCK_REALTIME), channel_state->shutdown_cq.cq(),
-          new TagSaver(nullptr));
-      grpc_channel_watch_connectivity_state(
-          channel_state->channel, channel_state->state,
-          gpr_inf_future(GPR_CLOCK_REALTIME), cq_.cq(),
-          new TagSaver(channel_state));
-    }
-  }
-
-  bool StopWatchingLocked() {
-    if (--channel_count_ == 0) {
-      {
-        std::unique_lock<std::mutex> lock(shutdown_mu_);
-        shutdown_ = true;
-        shutdown_cv_.notify_one();
-      }
-      gpr_thd_join(thd_id_);
-      return true;
-    }
-    return false;
-  }
-
-  friend void WatchStateChange(void* arg);
-  struct ChannelState {
-    explicit ChannelState(grpc_channel* channel)
-        : channel(channel), state(GRPC_CHANNEL_IDLE){};
-    grpc_channel* channel;
-    grpc_connectivity_state state;
-    CompletionQueue shutdown_cq;
-  };
-  gpr_thd_id thd_id_;
-  CompletionQueue cq_;
-  gpr_refcount ref_;
-  int channel_count_;
-
-  std::mutex shutdown_mu_;
-  std::condition_variable shutdown_cv_;  // protected by shutdown_mu_
-  bool shutdown_;                        // protected by shutdown_mu_
-
-  static std::mutex g_watcher_mu_;
-  static ChannelConnectivityWatcher* g_watcher_;  // protected by g_watcher_mu_
-};
-
-std::mutex ChannelConnectivityWatcher::g_watcher_mu_;
-ChannelConnectivityWatcher* ChannelConnectivityWatcher::g_watcher_ = nullptr;
-
-void WatchStateChange(void* arg) {
-  ChannelConnectivityWatcher* watcher =
-      static_cast<ChannelConnectivityWatcher*>(arg);
-  watcher->WatchStateChangeImpl();
-}
-}  // namespace
-
 static internal::GrpcLibraryInitializer g_gli_initializer;
 Channel::Channel(const grpc::string& host, grpc_channel* channel)
     : host_(host), c_channel_(channel) {
   g_gli_initializer.summon();
-  if (grpc_channel_support_connectivity_watcher(channel)) {
-    ChannelConnectivityWatcher::StartWatching(channel);
-  }
 }
 
-Channel::~Channel() {
-  const bool stop_watching =
-      grpc_channel_support_connectivity_watcher(c_channel_);
-  grpc_channel_destroy(c_channel_);
-  if (stop_watching) {
-    ChannelConnectivityWatcher::StopWatching();
-  }
-}
+Channel::~Channel() { grpc_channel_destroy(c_channel_); }
 
 namespace {
 
@@ -259,8 +85,9 @@ grpc::string Channel::GetServiceConfigJSON() const {
                              &channel_info.service_config_json);
 }
 
-Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
-                         CompletionQueue* cq) {
+internal::Call Channel::CreateCall(const internal::RpcMethod& method,
+                                   ClientContext* context,
+                                   CompletionQueue* cq) {
   const bool kRegistered = method.channel_tag() && context->authority().empty();
   grpc_call* c_call = NULL;
   if (kRegistered) {
@@ -292,10 +119,11 @@ Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
   }
   grpc_census_call_set_context(c_call, context->census_context());
   context->set_call(c_call, shared_from_this());
-  return Call(c_call, this, cq);
+  return internal::Call(c_call, this, cq);
 }
 
-void Channel::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
+void Channel::PerformOpsOnCall(internal::CallOpSetInterface* ops,
+                               internal::Call* call) {
   static const size_t MAX_OPS = 8;
   size_t nops = 0;
   grpc_op cops[MAX_OPS];
@@ -313,6 +141,24 @@ grpc_connectivity_state Channel::GetState(bool try_to_connect) {
   return grpc_channel_check_connectivity_state(c_channel_, try_to_connect);
 }
 
+namespace {
+
+class TagSaver final : public internal::CompletionQueueTag {
+ public:
+  explicit TagSaver(void* tag) : tag_(tag) {}
+  ~TagSaver() override {}
+  bool FinalizeResult(void** tag, bool* status) override {
+    *tag = tag_;
+    delete this;
+    return true;
+  }
+
+ private:
+  void* tag_;
+};
+
+}  // namespace
+
 void Channel::NotifyOnStateChangeImpl(grpc_connectivity_state last_observed,
                                       gpr_timespec deadline,
                                       CompletionQueue* cq, void* tag) {