Merge pull request #18476 from markdroth/c++_client_channel_external_watcher

Convert external connectivity watcher to C++.

src/core/ext/filters/client_channel/client_channel.cc

@@ -51,6 +51,7 @@
 #include "src/core/lib/gpr/string.h"
 #include "src/core/lib/gprpp/inlined_vector.h"
 #include "src/core/lib/gprpp/manual_constructor.h"
+#include "src/core/lib/gprpp/mutex_lock.h"
 #include "src/core/lib/iomgr/combiner.h"
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/lib/iomgr/polling_entity.h"
@@ -91,7 +92,55 @@ grpc_core::TraceFlag grpc_client_channel_routing_trace(
  * CHANNEL-WIDE FUNCTIONS
  */
 
-struct external_connectivity_watcher;
+// Forward declaration.
+typedef struct client_channel_channel_data channel_data;
+
+namespace grpc_core {
+namespace {
+
+class ExternalConnectivityWatcher {
+ public:
+  class WatcherList {
+   public:
+    WatcherList() { gpr_mu_init(&mu_); }
+    ~WatcherList() { gpr_mu_destroy(&mu_); }
+
+    int size() const;
+    ExternalConnectivityWatcher* Lookup(grpc_closure* on_complete) const;
+    void Add(ExternalConnectivityWatcher* watcher);
+    void Remove(const ExternalConnectivityWatcher* watcher);
+
+   private:
+    // head_ is guarded by a mutex, since the size() method needs to
+    // iterate over the list, and it's called from the C-core API
+    // function grpc_channel_num_external_connectivity_watchers(), which
+    // is synchronous and therefore cannot run in the combiner.
+    mutable gpr_mu mu_;
+    ExternalConnectivityWatcher* head_ = nullptr;
+  };
+
+  ExternalConnectivityWatcher(channel_data* chand, grpc_polling_entity pollent,
+                              grpc_connectivity_state* state,
+                              grpc_closure* on_complete,
+                              grpc_closure* watcher_timer_init);
+
+  ~ExternalConnectivityWatcher();
+
+ private:
+  static void OnWatchCompleteLocked(void* arg, grpc_error* error);
+  static void WatchConnectivityStateLocked(void* arg, grpc_error* ignored);
+
+  channel_data* chand_;
+  grpc_polling_entity pollent_;
+  grpc_connectivity_state* state_;
+  grpc_closure* on_complete_;
+  grpc_closure* watcher_timer_init_;
+  grpc_closure my_closure_;
+  ExternalConnectivityWatcher* next_ = nullptr;
+};
+
+}  // namespace
+}  // namespace grpc_core
 
 struct QueuedPick {
   LoadBalancingPolicy::PickArgs pick;
@@ -99,7 +148,7 @@ struct QueuedPick {
   QueuedPick* next = nullptr;
 };
 
-typedef struct client_channel_channel_data {
+struct client_channel_channel_data {
   bool deadline_checking_enabled;
   bool enable_retries;
   size_t per_rpc_retry_buffer_size;
@@ -139,11 +188,10 @@ typedef struct client_channel_channel_data {
   grpc_connectivity_state_tracker state_tracker;
   grpc_error* disconnect_error;
 
-  /* external_connectivity_watcher_list head is guarded by its own mutex, since
-   * counts need to be grabbed immediately without polling on a cq */
-  gpr_mu external_connectivity_watcher_list_mu;
-  struct external_connectivity_watcher* external_connectivity_watcher_list_head;
-} channel_data;
+  grpc_core::ManualConstructor<
+      grpc_core::ExternalConnectivityWatcher::WatcherList>
+      external_connectivity_watcher_list;
+};
 
 // Forward declarations.
 static void start_pick_locked(void* arg, grpc_error* ignored);
@@ -191,6 +239,123 @@ static void set_connectivity_state_and_picker_locked(
 namespace grpc_core {
 namespace {
 
+//
+// ExternalConnectivityWatcher::WatcherList
+//
+
+int ExternalConnectivityWatcher::WatcherList::size() const {
+  MutexLock lock(&mu_);
+  int count = 0;
+  for (ExternalConnectivityWatcher* w = head_; w != nullptr; w = w->next_) {
+    ++count;
+  }
+  return count;
+}
+
+ExternalConnectivityWatcher* ExternalConnectivityWatcher::WatcherList::Lookup(
+    grpc_closure* on_complete) const {
+  MutexLock lock(&mu_);
+  ExternalConnectivityWatcher* w = head_;
+  while (w != nullptr && w->on_complete_ != on_complete) {
+    w = w->next_;
+  }
+  return w;
+}
+
+void ExternalConnectivityWatcher::WatcherList::Add(
+    ExternalConnectivityWatcher* watcher) {
+  GPR_ASSERT(Lookup(watcher->on_complete_) == nullptr);
+  MutexLock lock(&mu_);
+  GPR_ASSERT(watcher->next_ == nullptr);
+  watcher->next_ = head_;
+  head_ = watcher;
+}
+
+void ExternalConnectivityWatcher::WatcherList::Remove(
+    const ExternalConnectivityWatcher* watcher) {
+  MutexLock lock(&mu_);
+  if (watcher == head_) {
+    head_ = watcher->next_;
+    return;
+  }
+  for (ExternalConnectivityWatcher* w = head_; w != nullptr; w = w->next_) {
+    if (w->next_ == watcher) {
+      w->next_ = w->next_->next_;
+      return;
+    }
+  }
+  GPR_UNREACHABLE_CODE(return );
+}
+
+//
+// ExternalConnectivityWatcher
+//
+
+ExternalConnectivityWatcher::ExternalConnectivityWatcher(
+    channel_data* chand, grpc_polling_entity pollent,
+    grpc_connectivity_state* state, grpc_closure* on_complete,
+    grpc_closure* watcher_timer_init)
+    : chand_(chand),
+      pollent_(pollent),
+      state_(state),
+      on_complete_(on_complete),
+      watcher_timer_init_(watcher_timer_init) {
+  grpc_polling_entity_add_to_pollset_set(&pollent_, chand_->interested_parties);
+  GRPC_CHANNEL_STACK_REF(chand_->owning_stack, "ExternalConnectivityWatcher");
+  GRPC_CLOSURE_SCHED(
+      GRPC_CLOSURE_INIT(&my_closure_, WatchConnectivityStateLocked, this,
+                        grpc_combiner_scheduler(chand_->combiner)),
+      GRPC_ERROR_NONE);
+}
+
+ExternalConnectivityWatcher::~ExternalConnectivityWatcher() {
+  grpc_polling_entity_del_from_pollset_set(&pollent_,
+                                           chand_->interested_parties);
+  GRPC_CHANNEL_STACK_UNREF(chand_->owning_stack, "ExternalConnectivityWatcher");
+}
+
+void ExternalConnectivityWatcher::OnWatchCompleteLocked(void* arg,
+                                                        grpc_error* error) {
+  ExternalConnectivityWatcher* self =
+      static_cast<ExternalConnectivityWatcher*>(arg);
+  grpc_closure* on_complete = self->on_complete_;
+  self->chand_->external_connectivity_watcher_list->Remove(self);
+  Delete(self);
+  GRPC_CLOSURE_SCHED(on_complete, GRPC_ERROR_REF(error));
+}
+
+void ExternalConnectivityWatcher::WatchConnectivityStateLocked(
+    void* arg, grpc_error* ignored) {
+  ExternalConnectivityWatcher* self =
+      static_cast<ExternalConnectivityWatcher*>(arg);
+  if (self->state_ == nullptr) {
+    // Handle cancellation.
+    GPR_ASSERT(self->watcher_timer_init_ == nullptr);
+    ExternalConnectivityWatcher* found =
+        self->chand_->external_connectivity_watcher_list->Lookup(
+            self->on_complete_);
+    if (found != nullptr) {
+      grpc_connectivity_state_notify_on_state_change(
+          &found->chand_->state_tracker, nullptr, &found->my_closure_);
+    }
+    Delete(self);
+    return;
+  }
+  // New watcher.
+  self->chand_->external_connectivity_watcher_list->Add(self);
+  // This assumes that the closure is scheduled on the ExecCtx scheduler
+  // and that GRPC_CLOSURE_RUN would run the closure immediately.
+  GRPC_CLOSURE_RUN(self->watcher_timer_init_, GRPC_ERROR_NONE);
+  GRPC_CLOSURE_INIT(&self->my_closure_, OnWatchCompleteLocked, self,
+                    grpc_combiner_scheduler(self->chand_->combiner));
+  grpc_connectivity_state_notify_on_state_change(
+      &self->chand_->state_tracker, self->state_, &self->my_closure_);
+}
+
+//
+// ClientChannelControlHelper
+//
+
 class ClientChannelControlHelper
     : public LoadBalancingPolicy::ChannelControlHelper {
  public:
@@ -402,12 +567,7 @@ static grpc_error* cc_init_channel_elem(grpc_channel_element* elem,
                                "client_channel");
   chand->disconnect_error = GRPC_ERROR_NONE;
   gpr_mu_init(&chand->info_mu);
-  gpr_mu_init(&chand->external_connectivity_watcher_list_mu);
-
-  gpr_mu_lock(&chand->external_connectivity_watcher_list_mu);
-  chand->external_connectivity_watcher_list_head = nullptr;
-  gpr_mu_unlock(&chand->external_connectivity_watcher_list_mu);
-
+  chand->external_connectivity_watcher_list.Init();
   chand->owning_stack = args->channel_stack;
   chand->deadline_checking_enabled =
       grpc_deadline_checking_enabled(args->channel_args);
@@ -515,7 +675,7 @@ static void cc_destroy_channel_elem(grpc_channel_element* elem) {
   GRPC_ERROR_UNREF(chand->disconnect_error);
   grpc_connectivity_state_destroy(&chand->state_tracker);
   gpr_mu_destroy(&chand->info_mu);
-  gpr_mu_destroy(&chand->external_connectivity_watcher_list_mu);
+  chand->external_connectivity_watcher_list.Destroy();
 }
 
 /*************************************************************************
@@ -2875,6 +3035,10 @@ const grpc_channel_filter grpc_client_channel_filter = {
     "client-channel",
 };
 
+//
+// functions exported to the rest of core
+//
+
 void grpc_client_channel_set_channelz_node(
     grpc_channel_element* elem, grpc_core::channelz::ClientChannelNode* node) {
   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
@@ -2914,120 +3078,10 @@ grpc_connectivity_state grpc_client_channel_check_connectivity_state(
   return out;
 }
 
-typedef struct external_connectivity_watcher {
-  channel_data* chand;
-  grpc_polling_entity pollent;
-  grpc_closure* on_complete;
-  grpc_closure* watcher_timer_init;
-  grpc_connectivity_state* state;
-  grpc_closure my_closure;
-  struct external_connectivity_watcher* next;
-} external_connectivity_watcher;
-
-static external_connectivity_watcher* lookup_external_connectivity_watcher(
-    channel_data* chand, grpc_closure* on_complete) {
-  gpr_mu_lock(&chand->external_connectivity_watcher_list_mu);
-  external_connectivity_watcher* w =
-      chand->external_connectivity_watcher_list_head;
-  while (w != nullptr && w->on_complete != on_complete) {
-    w = w->next;
-  }
-  gpr_mu_unlock(&chand->external_connectivity_watcher_list_mu);
-  return w;
-}
-
-static void external_connectivity_watcher_list_append(
-    channel_data* chand, external_connectivity_watcher* w) {
-  GPR_ASSERT(!lookup_external_connectivity_watcher(chand, w->on_complete));
-
-  gpr_mu_lock(&w->chand->external_connectivity_watcher_list_mu);
-  GPR_ASSERT(!w->next);
-  w->next = chand->external_connectivity_watcher_list_head;
-  chand->external_connectivity_watcher_list_head = w;
-  gpr_mu_unlock(&w->chand->external_connectivity_watcher_list_mu);
-}
-
-static void external_connectivity_watcher_list_remove(
-    channel_data* chand, external_connectivity_watcher* to_remove) {
-  GPR_ASSERT(
-      lookup_external_connectivity_watcher(chand, to_remove->on_complete));
-  gpr_mu_lock(&chand->external_connectivity_watcher_list_mu);
-  if (to_remove == chand->external_connectivity_watcher_list_head) {
-    chand->external_connectivity_watcher_list_head = to_remove->next;
-    gpr_mu_unlock(&chand->external_connectivity_watcher_list_mu);
-    return;
-  }
-  external_connectivity_watcher* w =
-      chand->external_connectivity_watcher_list_head;
-  while (w != nullptr) {
-    if (w->next == to_remove) {
-      w->next = w->next->next;
-      gpr_mu_unlock(&chand->external_connectivity_watcher_list_mu);
-      return;
-    }
-    w = w->next;
-  }
-  GPR_UNREACHABLE_CODE(return );
-}
-
 int grpc_client_channel_num_external_connectivity_watchers(
     grpc_channel_element* elem) {
   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
-  int count = 0;
-
-  gpr_mu_lock(&chand->external_connectivity_watcher_list_mu);
-  external_connectivity_watcher* w =
-      chand->external_connectivity_watcher_list_head;
-  while (w != nullptr) {
-    count++;
-    w = w->next;
-  }
-  gpr_mu_unlock(&chand->external_connectivity_watcher_list_mu);
-
-  return count;
-}
-
-static void on_external_watch_complete_locked(void* arg, grpc_error* error) {
-  external_connectivity_watcher* w =
-      static_cast<external_connectivity_watcher*>(arg);
-  grpc_closure* follow_up = w->on_complete;
-  grpc_polling_entity_del_from_pollset_set(&w->pollent,
-                                           w->chand->interested_parties);
-  GRPC_CHANNEL_STACK_UNREF(w->chand->owning_stack,
-                           "external_connectivity_watcher");
-  external_connectivity_watcher_list_remove(w->chand, w);
-  gpr_free(w);
-  GRPC_CLOSURE_SCHED(follow_up, GRPC_ERROR_REF(error));
-}
-
-static void watch_connectivity_state_locked(void* arg,
-                                            grpc_error* error_ignored) {
-  external_connectivity_watcher* w =
-      static_cast<external_connectivity_watcher*>(arg);
-  external_connectivity_watcher* found = nullptr;
-  if (w->state != nullptr) {
-    external_connectivity_watcher_list_append(w->chand, w);
-    // An assumption is being made that the closure is scheduled on the exec ctx
-    // scheduler and that GRPC_CLOSURE_RUN would run the closure immediately.
-    GRPC_CLOSURE_RUN(w->watcher_timer_init, GRPC_ERROR_NONE);
-    GRPC_CLOSURE_INIT(&w->my_closure, on_external_watch_complete_locked, w,
-                      grpc_combiner_scheduler(w->chand->combiner));
-    grpc_connectivity_state_notify_on_state_change(&w->chand->state_tracker,
-                                                   w->state, &w->my_closure);
-  } else {
-    GPR_ASSERT(w->watcher_timer_init == nullptr);
-    found = lookup_external_connectivity_watcher(w->chand, w->on_complete);
-    if (found) {
-      GPR_ASSERT(found->on_complete == w->on_complete);
-      grpc_connectivity_state_notify_on_state_change(
-          &found->chand->state_tracker, nullptr, &found->my_closure);
-    }
-    grpc_polling_entity_del_from_pollset_set(&w->pollent,
-                                             w->chand->interested_parties);
-    GRPC_CHANNEL_STACK_UNREF(w->chand->owning_stack,
-                             "external_connectivity_watcher");
-    gpr_free(w);
-  }
+  return chand->external_connectivity_watcher_list->size();
 }
 
 void grpc_client_channel_watch_connectivity_state(
@@ -3035,21 +3089,8 @@ void grpc_client_channel_watch_connectivity_state(
     grpc_connectivity_state* state, grpc_closure* closure,
     grpc_closure* watcher_timer_init) {
   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
-  external_connectivity_watcher* w =
-      static_cast<external_connectivity_watcher*>(gpr_zalloc(sizeof(*w)));
-  w->chand = chand;
-  w->pollent = pollent;
-  w->on_complete = closure;
-  w->state = state;
-  w->watcher_timer_init = watcher_timer_init;
-  grpc_polling_entity_add_to_pollset_set(&w->pollent,
-                                         chand->interested_parties);
-  GRPC_CHANNEL_STACK_REF(w->chand->owning_stack,
-                         "external_connectivity_watcher");
-  GRPC_CLOSURE_SCHED(
-      GRPC_CLOSURE_INIT(&w->my_closure, watch_connectivity_state_locked, w,
-                        grpc_combiner_scheduler(chand->combiner)),
-      GRPC_ERROR_NONE);
+  grpc_core::New<grpc_core::ExternalConnectivityWatcher>(
+      chand, pollent, state, closure, watcher_timer_init);
 }
 
 grpc_core::RefCountedPtr<grpc_core::SubchannelCall>