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

Code before:
    }
  }
}

ClusterUpdateCallbacksHandlePtr
ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;
  return cluster_manager.addClusterUpdateCallbacks(cb);
}

absl::StatusOr<OdCdsApiHandlePtr>
ClusterManagerImpl::allocateOdCdsApi(OdCdsCreationFunction creation_function,
                                     const envoy::config::core::v3::ConfigSource& odcds_config,
                                     OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,
                                     ProtobufMessage::ValidationVisitor& validation_visitor) {
  // Generate a unique key for this subscription based on config and locator.
  // This enables reuse of subscriptions with the same configuration.
  // Reference counting enables cleanup when the last handle is destroyed.
  uint64_t config_hash = MessageUtil::hash(odcds_config);
  if (odcds_resources_locator.has_value()) {
    config_hash = absl::HashOf(config_hash, MessageUtil::hash(*odcds_resources_locator));
  }

  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;
      }

Review comment (suggestion):
i think my question here was do we need to check return status of `handle` prior to invoking 
```
it->second->ref_count.fetch_add(1, std::memory_order_relaxed);
```

Code after:
    }
  }
}

ClusterUpdateCallbacksHandlePtr
ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;
  return cluster_manager.addClusterUpdateCallbacks(cb);
}

absl::StatusOr<OdCdsApiHandlePtr>
ClusterManagerImpl::allocateOdCdsApi(OdCdsCreationFunction creation_function,
                                     const envoy::config::core::v3::ConfigSource& odcds_config,
                                     OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,
                                     ProtobufMessage::ValidationVisitor& validation_visitor) {
  // Generate a unique key based on config and locator. This enables reuse of subscriptions
  // with the same configuration. Note that timeout is intentionally not part of the hash,
  // so different timeout values will share the same subscription.
  // Subscriptions persist for the lifetime of ClusterManagerImpl and are cleaned up when
  // it is destroyed.
  uint64_t config_hash = MessageUtil::hash(odcds_config);
  if (odcds_resources_locator.has_value()) {
    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 "