From 6a48dc544e4db970b19cf132019bcccb8eb92df1 Mon Sep 17 00:00:00 2001 From: Yuchen Zeng Date: Tue, 26 Sep 2017 19:49:23 -0700 Subject: [PATCH] Remove c++ channel connectivity watcher --- .../client_channel/connectivity_watcher.h | 2 + src/cpp/client/channel_cc.cc | 194 ++---------------- 2 files changed, 21 insertions(+), 175 deletions(-) diff --git a/src/core/ext/filters/client_channel/connectivity_watcher.h b/src/core/ext/filters/client_channel/connectivity_watcher.h index 89586dc7368..e12d6c284a1 100644 --- a/src/core/ext/filters/client_channel/connectivity_watcher.h +++ b/src/core/ext/filters/client_channel/connectivity_watcher.h @@ -23,6 +23,8 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/iomgr/exec_ctx.h" +/* Constantly watches client channel connectivity status to reconnect a + * transiently disconnected channel */ void grpc_client_channel_start_watching_connectivity( grpc_exec_ctx* exec_ctx, grpc_channel_element* client_channel_elem, grpc_channel_stack* channel_stack); diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc index 4049fc846b9..1509a92d268 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 lock(g_watcher_mu_); - if (g_watcher_ == nullptr) { - g_watcher_ = new ChannelConnectivityWatcher(); - } - g_watcher_->StartWatchingLocked(channel); - } - } - - static void StopWatching() { - if (!IsDisabled()) { - std::unique_lock 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 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(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 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(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 @@ 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 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) {