Remove c++ channel connectivity watcher
diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc
index 4049fc8..1509a92 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 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_;
-};
-
-// 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 {
@@ -315,6 +141,24 @@
return grpc_channel_check_connectivity_state(c_channel_, try_to_connect);
}
+namespace {
+
+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_;
+};
+
+} // namespace
+
void Channel::NotifyOnStateChangeImpl(grpc_connectivity_state last_observed,
gpr_timespec deadline,
CompletionQueue* cq, void* tag) {