Use WorkSerializer in XdsClient to propagate updates in a synchronized manner #27975

yashykt · 2021-11-09T07:42:33Z

XdsClient watcher notifications callbacks are currently invoked inline while still holding onto the XdsClient mutex. These callbacks often require to invoke an API on XdsClient. To do this in a way that avoids deadlocks, the callbacks schedule themselves on ExecCtx. The issue with this is that the callbacks can be pushed on ExecCtxs of different threads which can mess up the ordering of these notifications.

Using WorkSerializer::Schedule and WorkSerializer::DrainQueue, we always invoke the callbacks outside the XdsClient::mu_ so that the callbacks do not need to delay the callbacks. We additionally get the advantage of being able to execute code outside the critical region which should be a performance win (though it's probably not so critical).

This change is

yashykt · 2021-11-09T20:02:47Z

Mark is OOO, so adding @donnadionne for review. I probably won't merge this till I get Mark's eyes on this but it'll be good to have an additional set of eyes.

markdroth

This looks good overall! Comments are mostly optimizations, particularly around not taking unnecessary refs or making unnecessary copies.

Please let me know if you have any questions. Thanks!

Reviewed 5 of 6 files at r1, 1 of 1 files at r2, all commit messages.
Reviewable status: all files reviewed, 15 unresolved discussions (waiting on @donnadionne and @yashykt)

src/core/ext/filters/client_channel/lb_policy/xds/cds.cc, line 75 at r2 (raw file):

    void OnClusterChanged(XdsApi::CdsUpdate cluster_data) override {
      RefCountedPtr<CdsLb> cds_lb = parent_;

This can be a const reference, since we don't need to take a new ref here. The new ref will happen with the copy for the lambda capture.

Actually, since the watcher itself is now ref-counted, we can avoid copying name_ if we take a ref to the watcher instead of taking a ref to the CDS LB policy:

Ref().release();  // Ref held by callback.
parent_->work_serializer()->Run(
    // TODO(roth): When we move to C++14, capture cluster_data with std::move().
    [this, cluster_data]() mutable {
      parent_->OnClusterChanged(name_, std::move(cluster_data));
      Unref();
    },
    DEBUG_LOCATION);

Same for all three callbacks.

src/core/ext/filters/client_channel/lb_policy/xds/cds.cc, line 78 at r2 (raw file):

      std::string name = name_;
      parent_->work_serializer()->Run(
          [cds_lb, name, cluster_data]() mutable {

Please add a TODO about capturing cluster_data via std::move() once we can use C++14.

src/core/ext/filters/client_channel/lb_policy/xds/xds_cluster_resolver.cc, line 148 at r2 (raw file):

    }

    // parent() and index() could have been protected methods but a bug in msvc

If possible, I'd prefer to keep these methods protected and use friend statements to work around this bug, like we were doing before. If MSVC doesn't understand the lambdas, then maybe we can move the code from the lambdas to private methods of EndpointWatcher. For example:

class EndpointWatcher : public XdsClient::EndpointWatcherInterface {
 public:
  void OnEndpointChanged(XdsApi::EdsUpdate update) override {
    Ref().release();  // Ref held by callback.
    discovery_mechanism_->parent()->work_serializer()->Run(
        [this, update]() mutable {
          OnEndpointChangedHelper(std::move(update));
          Unref();
        },
        DEBUG_LOCATION);
  }

 private:
  // Code accessing protected methods of `DiscoveryMechanism` need to be
  // in methods of this class rather than in lambdas to work around an MSVC bug.
  void OnEndpointChangedHelper(XdsApi::EdsUpdate update) {
    discovery_mechanism_->parent()->OnEndpointChanged(
        discovery_mechanism_->index(), std::move(update));
  }
};

src/core/ext/filters/client_channel/lb_policy/xds/xds_cluster_resolver.cc, line 181 at r2 (raw file):

      }
      void OnEndpointChanged(XdsApi::EdsUpdate update) override {
        RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism =

This can be a const reference, so that we don't take an unnecessary ref. We are already taking a ref in the lambda capture.

Same thing for all three callbacks.

(This becomes irrelevant if you take my suggestion above about taking a new ref to the watcher instead of taking a ref to the discovery mechanism to avoid the MSVC bug.)

src/core/ext/filters/client_channel/lb_policy/xds/xds_cluster_resolver.cc, line 184 at r2 (raw file):

            discovery_mechanism_;
        discovery_mechanism->parent()->work_serializer()->Run(
            [discovery_mechanism, update]() mutable {

Please add a TODO about capturing update via std::move() once we can use C++14.

src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc, line 899 at r2 (raw file):

void XdsResolver::NotifyLdsUpdate(XdsApi::LdsUpdate update,
                                  const DebugLocation& location) {
  RefCountedPtr<XdsResolver> resolver = Ref();

This is taking two refs, one here and another in the lambda capture, but we only need one.

I think it makes sense to keep this code in the watcher implementations rather than moving it into these helper functions, because then we can avoid the extra ref by taking the same approach as elsewhere (i.e., take a ref to the watcher), or we can declare a const reference to the existing RefCountedPtr<XdsResolver> data member.

Same thing for all three callbacks.

src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc, line 901 at r2 (raw file):

  RefCountedPtr<XdsResolver> resolver = Ref();
  work_serializer_->Run(
      [resolver, update]() mutable {

Please add a TODO about using std::move() to capture update when we move to C++14.

Same for the RDS update.

src/core/ext/xds/xds_client.h, line 49 at r2 (raw file):

  class ListenerWatcherInterface : public RefCounted<ListenerWatcherInterface> {
   public:
    // TODO(review_note): Tried adding ABSL_EXCLUSIVE_LOCKS_REQUIRED but that

What was the error you got when trying this?

Note that when using lock annotations with a WorkSerializer, you need to add ABSL_EXCLUSIVE_LOCKS_REQUIRED() annotations to the lambdas that you run in the WorkSerializer. You can see how we did this in the client channel code in #25983.

src/core/ext/xds/xds_client.cc, line 183 at r2 (raw file):

  class ResourceState : public InternallyRefCounted<ResourceState> {
   public:
    // TODO(): Check if passing XdsClient as a raw ptr is safe

I don't think we need to pass this in at all. Instead, you can change OnTimer() to access it via self->ads_calld_->xds_client().

src/core/ext/xds/xds_client.cc, line 226 at r2 (raw file):

        self->OnTimerLocked(GRPC_ERROR_REF(error));
      }
      // TODO(): Can we avoid draining the queue on a timer thread?

I think this is fine. We were already doing this work in the timer thread before this change.

src/core/ext/xds/xds_client.cc, line 2162 at r2 (raw file):

        "Unable to parse resource name for listener %s", listener_name));
    work_serializer_.Run(
        [watcher, error]() {

Please add a TODO that once we can use C++14, we can capture watcher using std::move().

Same thing for all resource types.

src/core/ext/xds/xds_client.cc, line 2163 at r2 (raw file):

    work_serializer_.Run(
        [watcher, error]() {
          // review_note: this should fix a leak

Good catch!

Please remove these review_note: comments before merging.

src/core/ext/xds/xds_client.cc, line 2184 at r2 (raw file):

      auto& value = listener_state.update.value();
      work_serializer_.Schedule(
          [watcher, value]() { watcher->OnListenerChanged(value); },

Same C++14 TODO here.

Same thing for all resource types.

src/core/ext/xds/xds_client.cc, line 2589 at r2 (raw file):

  // being captured by value
  work_serializer_.Schedule(
      [authority_state_map, error]() ABSL_NO_THREAD_SAFETY_ANALYSIS {

I don't think we want to capture the whole authority_state_map here, since that includes a lot of data we don't actually need. It will take an unnecessary ref to the ChannelState object, and it will unnecessarily copy all of the cached resource data.

Instead, I suggest something like this:

std::set<RefCountedPtr<ListenerWatcher>> listener_watchers;
std::set<RefCountedPtr<RouteConfigWatcher>> route_config_watchers;
std::set<RefCountedPtr<ClusterWatcher>> cluster_watchers;
std::set<RefCountedPtr<EndpointWatcher>> endpoint_watchers;
for (const auto& a : authority_state_map_) {
  for (const auto& p : a.second.listener_map) {
    const ListenerState& listener_state = p.second;
    for (const auto& q : listener_state.watchers) {
      listener_watchers.insert(q.second);
    }
  }
  // ...same for other resource types...
}
work_serializer_.Schedule(
    [listener_watchers, route_config_watchers, cluster_watchers, endpoint_watchers, error]()
        ABSL_EXCLUSIVE_LOCKS_REQUIRED(&work_serializer_) {
      for (const auto& watcher : listener_watchers) {
        watcher->OnError(GRPC_ERROR_REF(error);
      }
      // ...same for other resource types...
      GRPC_ERROR_UNREF(error);
    },
    DEBUG_LOCATION);

src/core/ext/xds/xds_client.cc, line 2589 at r2 (raw file):

  // being captured by value
  work_serializer_.Schedule(
      [authority_state_map, error]() ABSL_NO_THREAD_SAFETY_ANALYSIS {

This should use ABSL_EXCLUSIVE_LOCKS_REQUIRED(&work_serializer_) instead of ABSL_NO_THREAD_SAFETY_ANALYSIS.

yashykt

Reviewable status: 1 of 6 files reviewed, 15 unresolved discussions (waiting on @donnadionne and @markdroth)

src/core/ext/filters/client_channel/lb_policy/xds/cds.cc, line 75 at r2 (raw file):