Repository: envoyproxy/envoy
File: source/common/upstream/cluster_manager_impl.cc (C++)

Code before:

  auto it = odcds_subscriptions_.find(config_hash);
  if (it != odcds_subscriptions_.end()) {
    auto handle = OdCdsApiHandleImpl::create(*this, config_hash);
    it->second->ref_count.fetch_add(1, std::memory_order_relaxed);
    return handle;
  }

  auto odcds_or_error =
      creation_function(odcds_config, odcds_resources_locator, xds_manager_, *this, *this,
                        *stats_.rootScope(), validation_visitor, context_);
  RETURN_IF_NOT_OK_REF(odcds_or_error.status());
  odcds_subscriptions_.emplace(
      config_hash, std::make_unique<OdCdsSubscriptionEntry>(std::move(*odcds_or_error)));
  return OdCdsApiHandleImpl::create(*this, config_hash);
}

ClusterManagerImpl::OdCdsApiHandleImpl::~OdCdsApiHandleImpl() {
  if (parent_.shutdown_.load()) {
    return;
  }

  if (parent_.dispatcher_.isThreadSafe()) {
    parent_.releaseOdCdsSubscription(subscription_key_);
  } else {
    parent_.dispatcher_.post([subscription_key = subscription_key_, &parent = parent_] {
      if (parent.shutdown_.load()) {
        return;
      }
      parent.releaseOdCdsSubscription(subscription_key);
    });
  }
}

void ClusterManagerImpl::releaseOdCdsSubscription(uint64_t subscription_key) {
  auto it = odcds_subscriptions_.find(subscription_key);
  if (it == odcds_subscriptions_.end()) {
    return;
  }

  if (it->second->ref_count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
    ENVOY_LOG(debug, "cm odcds: removing subscription with key {} (no more references)",
              subscription_key);
    odcds_subscriptions_.erase(it);
  }
}

ClusterDiscoveryCallbackHandlePtr
ClusterManagerImpl::requestOnDemandClusterDiscovery(uint64_t subscription_key, std::string name,
                                                    ClusterDiscoveryCallbackPtr callback,
                                                    std::chrono::milliseconds timeout) {

Review comment (performance):
I'm worried about the following race:
1. worker thread1 adds an OD-CDS subscription (following a request), and initiates a connection to the server.
2. server returns a cluster.
3. (the above can happen to multiple clusters)
4. an RDS update arrives and replaces the OD-CDS with the same contents
5. AFAICT the OD-CDS subscription is removed, which causes the removal of all the clusters that arrived in step 2.
6. Same requests that were working previously will now require new OD-CDS update.

I think a better design of the lifetime of an OD-CDS config and clusters received from the non-ADS config sources should be made before trying to optimize the lifetime of these subscriptions.

Code after:
    config_hash = absl::HashOf(config_hash, MessageUtil::hash(*odcds_resources_locator));
  }

  auto it = odcds_subscriptions_.find(config_hash);
  if (it != odcds_subscriptions_.end()) {
    return OdCdsApiHandleImpl::create(*this, config_hash);
  }

  auto odcds_or_error =
      creation_function(odcds_config, odcds_resources_locator, xds_manager_, *this, *this,
                        *stats_.rootScope(), validation_visitor, context_);
  RETURN_IF_NOT_OK_REF(odcds_or_error.status());
  odcds_subscriptions_.emplace(config_hash, std::move(*odcds_or_error));
  return OdCdsApiHandleImpl::create(*this, config_hash);
}

ClusterDiscoveryCallbackHandlePtr
ClusterManagerImpl::requestOnDemandClusterDiscovery(uint64_t config_source_key, std::string name,
                                                    ClusterDiscoveryCallbackPtr callback,
                                                    std::chrono::milliseconds timeout) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;

  auto [handle, discovery_in_progress, invoker] =
      cluster_manager.cdm_.addCallback(name, std::move(callback));
  // This check will catch requests for discoveries from this thread only. If other thread
  // requested the same discovery, we will detect it in the main thread later.
  if (discovery_in_progress) {
    ENVOY_LOG(debug,
              "cm odcds: on-demand discovery for cluster {} is already in progress, something else "
              "in thread {} has already requested it",
              name, cluster_manager.thread_local_dispatcher_.name());
    // This worker thread has already requested a discovery of a cluster with this name, so
    // nothing more left to do here.
    //
    // We can't "just" return handle here, because handle is a part of the structured binding done
    // above. So it's not really a ClusterDiscoveryCallbackHandlePtr, but more like
    // ClusterDiscoveryCallbackHandlePtr&, so named return value optimization does not apply here
    // - it needs to be moved.
    return std::move(handle);
  }
  ENVOY_LOG(
      debug,
      "cm odcds: forwarding the on-demand discovery request for cluster {} to the main thread",
      name);
  // This seems to be the first request for discovery of this cluster in this worker thread. Rest
  // of the process may only happen in the main thread.
  Event::Dispatcher& worker_dispatcher = cluster_manager.thread_local_dispatcher_;
  dispatcher_.post([this, config_source_key, timeout, name = std::move(name),
                    invoker = std::move(invoker), &worker_dispatcher] {
    OdCdsApiSharedPtr odcds = odcds_subscriptions_.at(config_source_key);